perf(gfql): cardinality-aware lean combine for seeded chains (#1755)

CHANGELOG.md

@@ -8,6 +8,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ## [Development]
 <!-- Do Not Erase This Section - Used for tracking unreleased changes -->
 
+### Performance
+- **GFQL seeded-chain lean combine (#1755)**: The two-pass chain executor (`_chain_impl` backward pass + `combine_steps`) reconciled wavefront boundaries with full-node-frame `safe_merge`/`pandas.merge` calls even when the seeded result was a single row, so a seeded 1-hop paid a whole-frame join. When the small side is at least `4x` smaller than the full frame (and is unique on the id with no extra columns), the byte-identical result is now obtained by an `isin` membership filter instead of the join machinery. Gated narrowly on cardinality/uniqueness/column-set and pandas-only (cuDF's GPU hash-join is already sub-ms; polars takes its own engine path); `GFQL_LEAN_COMBINE=0` restores the legacy merges. Parity is byte-identical (lean-on vs lean-off) across seeded node lookup, seeded 1-hop expand, and the native typed chain, including a null-key equivalence guard.
+
 ### Documentation
 - **GFQL engine-selection docs (pandas / polars / cuDF / polars-gpu)**: New :doc:`Choosing a GFQL Engine <gfql/engines>` page — a numbers-first, persona-tested guide to the four interchangeable engines. Adds the one-keyword `engine='polars'` speedup (up to ~38× over pandas on real graphs, no GPU), a motivating warm-median comparison table on real public graphs (LiveJournal 35M / Orkut 117M), a decision matrix (workload shape × size × hardware → engine, with the measured ~10K-edge CPU crossover, GPU-work-bound rule, polars-gpu memory-pressure caveat, and GPU-or-error contract), a cuDF-vs-polars-gpu disambiguation (eager-op vs fused-lazy; cuDF is not deprecated), an honest "when *not* to use Polars" section, the differential-parity guarantee, and a methodology + reproducer-script disclosure. Rewrote the top of `gfql/performance.rst` to lead with the engine comparison (de-marketed the prose), wired the new page into the GFQL toctree + recommended paths, and added Polars/polars-gpu to the engine examples in `gfql/quick.rst` and `gfql/about.rst` (previously only pandas/cuDF were documented). Driven by 4-persona doc user-testing (pandas DS, RAPIDS user, perf engineer, skeptical evaluator).
 

graphistry/compute/chain.py

@@ -9,7 +9,7 @@
 from graphistry.util import setup_logger
 from graphistry.utils.json import JSONVal
 from .ast import ASTObject, ASTNode, ASTEdge, from_json as ASTObject_from_json, serialize_binding_ops
-from .typing import DataFrameT
+from .typing import DataFrameT, SeriesT
 from .util import generate_safe_column_name
 from graphistry.compute.validate.validate_schema import validate_chain_schema
 from graphistry.compute.gfql.same_path_types import (
@@ -37,6 +37,12 @@ def _filter_edges_by_endpoint(
     return edges_df[edges_df[edge_col].isin(nodes_df[node_id])]
 
 
+from .chain_lean_combine import (
+    _lean_combine_enabled, _is_unique_ids, _lean_engine_ok,
+    _lean_intersect_full, _lean_prefilter_right,
+)
+
+
 class Chain(ASTSerializable):
 
     def __init__(
@@ -413,7 +419,12 @@ def apply_output_slice(op: ASTObject, op_label: ASTObject, df):
                 out_df = out_df[~has_na | has_tag]
 
     g_df = getattr(g, df_fld)
-    out_df = safe_merge(out_df, g_df, on=id, how='left', engine=engine)
+    # slice 5 (#1755): a seeded result attaches the full node/edge frame via a
+    # how='left' merge whose big side (g_df) is scanned in full even for a 1-row
+    # out_df. Pre-shrink g_df to the ids actually present (unmatched rows are
+    # dropped by how='left' regardless) so the join runs small-vs-small.
+    g_df_join = _lean_prefilter_right(out_df, g_df, id, engine) if id in out_df.columns else g_df
+    out_df = safe_merge(out_df, g_df_join, on=id, how='left', engine=engine)
 
     if logger.isEnabledFor(logging.DEBUG):
         logger.debug('COMBINED[%s] >> %s', kind, dbg_df(out_df))
@@ -1050,7 +1061,12 @@ def _chain_impl(
                     prev_orig_step = g_stack[-(len(g_stack_reverse) + 2)]
                 prev_wavefront_nodes = prev_loop_step._nodes
                 if g._node is not None and prev_wavefront_nodes is not None and g._nodes is not None:
-                    prev_wavefront_nodes = safe_merge(
+                    # slice 5 (#1755): re-attach full node columns to the reverse
+                    # wavefront. The inner merge scans all of g._nodes; when the
+                    # wavefront is tiny (seeded chain), the byte-identical result
+                    # is an isin membership filter (see _lean_intersect_full).
+                    _lean = _lean_intersect_full(g._nodes, prev_wavefront_nodes[[g._node]], g._node, engine_concrete)
+                    prev_wavefront_nodes = _lean if _lean is not None else safe_merge(
                         g._nodes,
                         prev_wavefront_nodes[[g._node]],
                         on=g._node,
@@ -1059,7 +1075,8 @@ def _chain_impl(
                     )
                 target_wave_front_nodes = prev_orig_step._nodes if prev_orig_step is not None else None
                 if g._node is not None and target_wave_front_nodes is not None and g._nodes is not None:
-                    target_wave_front_nodes = safe_merge(
+                    _lean = _lean_intersect_full(g._nodes, target_wave_front_nodes[[g._node]], g._node, engine_concrete)
+                    target_wave_front_nodes = _lean if _lean is not None else safe_merge(
                         g._nodes,
                         target_wave_front_nodes[[g._node]],
                         on=g._node,

graphistry/compute/chain_lean_combine.py

@@ -0,0 +1,107 @@
+"""Cardinality-aware lean-combine DataFrameT helpers for the seeded chain executor (#1755).
+
+Extracted from chain.py to keep the executor readable (mirrors the gfql_fast_paths.py
+#1731 convention: one-directional import, no back-edge). These are byte-identical
+replacements for the two-pass chain executor's full-frame `safe_merge`/`pandas.merge`
+reconciliations when the small side is much smaller + unique on the id — an `isin`
+membership filter instead of the join machinery. pandas-only (see `_lean_engine_ok`);
+`GFQL_LEAN_COMBINE=0` restores the legacy merges. Imports only leaf modules.
+"""
+import os as _os
+from typing import Optional
+from graphistry.Engine import Engine
+from graphistry.compute.typing import DataFrameT, SeriesT
+
+
+def _lean_combine_enabled() -> bool:
+    """Fast path is on by default; set GFQL_LEAN_COMBINE=0 to force the legacy
+    full-frame merges (used by the differential parity harness)."""
+    return _os.environ.get('GFQL_LEAN_COMBINE', '1') != '0'
+
+
+# Only engage when the small side is at least this many times smaller than the
+# full frame — below the gate the isin pre-pass is pure overhead vs the merge.
+_LEAN_SHRINK_RATIO = 4
+
+
+def _is_unique_ids(ids: SeriesT) -> bool:
+    """Cheap uniqueness probe on a (small) id Series, engine-agnostic."""
+    try:
+        n = len(ids)
+        if n <= 1:
+            return True
+        # pandas/cuDF both expose is_unique; fall back to nunique.
+        iu = getattr(ids, 'is_unique', None)
+        if iu is not None:
+            return bool(iu)
+        return int(ids.nunique()) == n
+    except Exception:
+        return False
+
+
+def _lean_engine_ok(engine: Engine) -> bool:
+    """Only pandas benefits: pandas' merge machinery (join-indexer build + block
+    consolidation) dominates the seeded chain, so an isin pre-filter is a large
+    win. On cuDF the GPU hash-join is already sub-ms and the extra isin pass +
+    boolean-mask gather is net-negative (measured dgx-spark 26.02, ~0.95-0.99x),
+    so the fast path stays off there. polars/dask/spark take their own engines."""
+    return engine == Engine.PANDAS
+
+
+def _lean_intersect_full(full: DataFrameT, key_frame: DataFrameT, key: str, engine: Engine) -> Optional[DataFrameT]:
+    """Byte-identical replacement for ``safe_merge(full, key_frame[[key]], on=key,
+    how='inner')`` when ``key_frame`` carries only ``key`` and is small + unique.
+
+    ``full`` is the whole node frame (unique on ``key``); the inner merge keeps
+    exactly the ``full`` rows whose id is in ``key_frame`` (1:1, ``full`` order,
+    ``full`` columns), which is precisely an isin filter. Returns ``None`` to
+    signal "not applicable — use the merge" (guards fan-out / column carry).
+    """
+    if not _lean_combine_enabled() or not _lean_engine_ok(engine):
+        return None
+    if key not in full.columns or key not in key_frame.columns:
+        return None
+    try:
+        full_len = len(full)
+        small_len = len(key_frame)
+    except Exception:
+        return None
+    if small_len == 0:
+        return None
+    if small_len * _LEAN_SHRINK_RATIO > full_len:
+        return None  # not enough size gap to be worth the isin pass
+    # key_frame must contribute no columns beyond key, and be unique on key, else
+    # the inner merge would fan out / add columns and diverge from an isin filter.
+    if list(key_frame.columns) != [key]:
+        return None
+    if not _is_unique_ids(key_frame[key]):
+        return None
+    # isin() matches nulls, but merge's null-key semantics are version-dependent;
+    # only the (small) key_frame need be null-free for equivalence: if it carries
+    # no null, both isin and inner-merge drop full's null-id rows identically,
+    # regardless of nulls in full. Checking the small side keeps this O(small).
+    if bool(key_frame[key].isna().any()):
+        return None
+    out = full[full[key].isin(key_frame[key])]
+    # match the merge's RangeIndex so any positional downstream use is identical.
+    return out.reset_index(drop=True)
+
+
+def _lean_prefilter_right(left: DataFrameT, right: DataFrameT, key: str, engine: Engine) -> DataFrameT:
+    """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.
+    """
+    if not _lean_combine_enabled() or not _lean_engine_ok(engine):
+        return right
+    if key not in left.columns or key not in right.columns:
+        return right
+    try:
+        left_len = len(left)
+        right_len = len(right)
+    except Exception:
+        return right
+    if left_len == 0 or left_len * _LEAN_SHRINK_RATIO > right_len:
+        return right
+    return right[right[key].isin(left[key])]

graphistry/tests/compute/test_chain_lean_combine.py

@@ -0,0 +1,202 @@
+"""Parity + gate tests for the cardinality-aware "lean combine" (#1755, slice 5).
+
+The lean path replaces full-node-frame ``safe_merge`` reconciliations in the
+two-pass chain executor with ``isin`` membership filters when the wavefront is
+much smaller than the full frame. It is byte-identical by construction; these
+tests pin that (lean-on vs lean-off, via ``GFQL_LEAN_COMBINE``) and exercise the
+narrow applicability gate directly.
+"""
+import os
+
+import numpy as np
+import pandas as pd
+import pytest
+
+import graphistry
+from graphistry.compute.ast import n, e_forward, e_reverse
+from graphistry.compute.chain import (
+    _is_unique_ids,
+    _lean_combine_enabled,
+    _lean_engine_ok,
+    _lean_intersect_full,
+    _lean_prefilter_right,
+)
+from graphistry.Engine import Engine
+
+
+@pytest.fixture(autouse=True)
+def _restore_lean_env():
+    """Keep GFQL_LEAN_COMBINE hermetic across tests."""
+    prev = os.environ.get('GFQL_LEAN_COMBINE')
+    yield
+    if prev is None:
+        os.environ.pop('GFQL_LEAN_COMBINE', None)
+    else:
+        os.environ['GFQL_LEAN_COMBINE'] = prev
+
+
+def _seeded_graph(n_persons: int = 1000, n_messages: int = 4000, seed: int = 0):
+    """Message -> Person HAS_CREATOR graph (same shape as the #1755 probe)."""
+    rng = np.random.default_rng(seed)
+    persons = pd.DataFrame({"id": np.arange(n_persons), "type": "Person"})
+    messages = pd.DataFrame(
+        {"id": np.arange(n_persons, n_persons + n_messages), "type": "Message"}
+    )
+    ndf = pd.concat([persons, messages], ignore_index=True)
+    edf = pd.DataFrame({
+        "src": np.arange(n_persons, n_persons + n_messages),
+        "dst": rng.integers(0, n_persons, n_messages),
+        "type": "HAS_CREATOR",
+    })
+    return graphistry.nodes(ndf, "id").edges(edf, "src", "dst"), n_persons
+
+
+# ---------------------------------------------------------------------------
+# gate unit tests
+# ---------------------------------------------------------------------------
+
+def test_lean_combine_enabled_env():
+    os.environ.pop('GFQL_LEAN_COMBINE', None)
+    assert _lean_combine_enabled() is True          # default on
+    os.environ['GFQL_LEAN_COMBINE'] = '0'
+    assert _lean_combine_enabled() is False
+    os.environ['GFQL_LEAN_COMBINE'] = '1'
+    assert _lean_combine_enabled() is True
+
+
+def test_lean_engine_ok_pandas_only():
+    assert _lean_engine_ok(Engine.PANDAS) is True
+    assert _lean_engine_ok(Engine.CUDF) is False
+    assert _lean_engine_ok(Engine.DASK) is False
+
+
+def test_is_unique_ids():
+    assert _is_unique_ids(pd.Series([], dtype='int64')) is True
+    assert _is_unique_ids(pd.Series([7])) is True
+    assert _is_unique_ids(pd.Series([1, 2, 3])) is True
+    assert _is_unique_ids(pd.Series([1, 2, 2])) is False
+
+
+def test_lean_intersect_full_matches_inner_merge():
+    full = pd.DataFrame({"id": np.arange(1000), "val": np.arange(1000) * 2})
+    key_frame = pd.DataFrame({"id": [3, 900, 12]})
+    lean = _lean_intersect_full(full, key_frame, "id", Engine.PANDAS)
+    merged = full.merge(key_frame[["id"]], on="id", how="inner").reset_index(drop=True)
+    assert lean is not None
+    pd.testing.assert_frame_equal(lean, merged)
+
+
+def test_lean_intersect_full_declines_when_not_applicable():
+    full = pd.DataFrame({"id": np.arange(100), "val": np.arange(100)})
+    # small side not >=4x smaller -> decline
+    big_key = pd.DataFrame({"id": np.arange(50)})
+    assert _lean_intersect_full(full, big_key, "id", Engine.PANDAS) is None
+    # key_frame carries an extra column -> would fan-out/add cols -> decline
+    extra = pd.DataFrame({"id": [1, 2], "extra": [9, 9]})
+    assert _lean_intersect_full(full, extra, "id", Engine.PANDAS) is None
+    # non-unique key -> decline
+    dup = pd.DataFrame({"id": [1, 1]})
+    assert _lean_intersect_full(full, dup, "id", Engine.PANDAS) is None
+    # non-pandas engine -> decline
+    assert _lean_intersect_full(full, pd.DataFrame({"id": [1]}), "id", Engine.CUDF) is None
+    # null in the (small) key frame -> decline (isin matches nulls, merge is
+    # version-dependent; declining keeps the equivalence version-independent)
+    nullkey = pd.DataFrame({"id": [1.0, np.nan]})
+    assert _lean_intersect_full(full.astype({"id": float}), nullkey, "id", Engine.PANDAS) is None
+    # disabled -> decline
+    os.environ['GFQL_LEAN_COMBINE'] = '0'
+    assert _lean_intersect_full(full, pd.DataFrame({"id": [1]}), "id", Engine.PANDAS) is None
+
+
+def test_lean_prefilter_right_matches_left_merge():
+    left = pd.DataFrame({"id": [5, 10]})
+    right = pd.DataFrame({"id": np.arange(1000), "val": np.arange(1000)})
+    shrunk = _lean_prefilter_right(left, right, "id", Engine.PANDAS)
+    # left merge result identical whether right is pre-shrunk or not
+    full_merge = left.merge(right, on="id", how="left")
+    lean_merge = left.merge(shrunk, on="id", how="left")
+    pd.testing.assert_frame_equal(full_merge, lean_merge)
+    assert len(shrunk) <= len(right)
+
+
+def test_lean_prefilter_right_noop_when_left_not_smaller():
+    left = pd.DataFrame({"id": np.arange(50)})
+    right = pd.DataFrame({"id": np.arange(100), "val": np.arange(100)})
+    out = _lean_prefilter_right(left, right, "id", Engine.PANDAS)
+    assert out is right  # untouched
+
+
+# ---------------------------------------------------------------------------
+# end-to-end parity: lean-on vs lean-off must be byte-identical
+# ---------------------------------------------------------------------------
+
+def _run(chain_ops, lean: str):
+    os.environ['GFQL_LEAN_COMBINE'] = lean
+    g, _ = _seeded_graph()
+    out = g.gfql(chain_ops, engine='pandas')
+    return (
+        out._nodes.sort_values('id').reset_index(drop=True),
+        out._edges.sort_values(['src', 'dst']).reset_index(drop=True),
+    )
+
+
+@pytest.mark.parametrize("ops_name", ["is5_creator", "is5_typed", "expand_both"])
+def test_seeded_chain_parity_lean_on_off(ops_name):
+    seed_msg = 1000 + 456  # a Message id (n_persons=1000)
+    ops = {
+        "is5_creator": [n({"id": seed_msg}), e_forward(), n()],
+        "is5_typed": [
+            n({"id": seed_msg}),
+            e_forward(edge_match={"type": "HAS_CREATOR"}),
+            n({"type": "Person"}),
+        ],
+        "expand_both": [n({"id": seed_msg}), e_forward(), n({"type": "Person"})],
+    }[ops_name]
+    on_nodes, on_edges = _run(ops, lean='1')
+    off_nodes, off_edges = _run(ops, lean='0')
+    pd.testing.assert_frame_equal(on_nodes, off_nodes)
+    pd.testing.assert_frame_equal(on_edges, off_edges)
+
+
+def test_lean_path_actually_engages(monkeypatch):
+    """Guard against a vacuous parity test: on a seeded chain the lean intersect
+    must fire at least once (return a non-None frame)."""
+    import graphistry.compute.chain as chain_mod
+
+    real_intersect = chain_mod._lean_intersect_full
+    real_prefilter = chain_mod._lean_prefilter_right
+    hits = {"intersect": 0, "prefilter": 0}
+
+    def _spy_intersect(full, key_frame, key, engine):
+        out = real_intersect(full, key_frame, key, engine)
+        if out is not None:
+            hits["intersect"] += 1
+        return out
+
+    def _spy_prefilter(left, right, key, engine):
+        out = real_prefilter(left, right, key, engine)
+        # engaged iff it actually shrank the right frame
+        try:
+            if len(out) < len(right):
+                hits["prefilter"] += 1
+        except Exception:
+            pass
+        return out
+
+    monkeypatch.setattr(chain_mod, "_lean_intersect_full", _spy_intersect)
+    monkeypatch.setattr(chain_mod, "_lean_prefilter_right", _spy_prefilter)
+    os.environ['GFQL_LEAN_COMBINE'] = '1'
+    g, n_persons = _seeded_graph()
+    seed_msg = n_persons + 456
+    # A MULTI-hop seeded chain with a backward reconciliation: the single-hop
+    # degenerate fast path (#1755) intercepts a seeded 1-hop before combine_steps,
+    # so lean-combine (which optimizes combine_steps + the backward pass) is
+    # exercised by a chain that actually reaches those passes. fwd->typed->rev
+    # engages BOTH lean helpers (combine intersect + backward-pass prefilter).
+    g.gfql(
+        [n({"id": seed_msg}), e_forward(), n({"type": "Person"}), e_reverse(), n()],
+        engine='pandas',
+    )
+    # both lean paths must fire on a seeded chain (else parity tests are vacuous)
+    assert hits["intersect"] >= 1
+    assert hits["prefilter"] >= 1