diff --git a/CHANGELOG.md b/CHANGELOG.md index 729502ef07..fe8191113e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ## [Development] +### 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 ` 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). diff --git a/graphistry/compute/chain.py b/graphistry/compute/chain.py index b1f316bde1..ccf4d3f349 100644 --- a/graphistry/compute/chain.py +++ b/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, diff --git a/graphistry/compute/chain_lean_combine.py b/graphistry/compute/chain_lean_combine.py new file mode 100644 index 0000000000..311f0dd9c0 --- /dev/null +++ b/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])] diff --git a/graphistry/tests/compute/test_chain_lean_combine.py b/graphistry/tests/compute/test_chain_lean_combine.py new file mode 100644 index 0000000000..d713b9d8c9 --- /dev/null +++ b/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