Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm

### Fixed
- **GFQL polars engine natively runs `UNWIND` of a carried `collect()` list column**: `... WITH collect(x) AS xs UNWIND xs AS y RETURN ...` (exploding a list-valued column produced by `collect()`, the row-pipeline analogue of the IC6 unwind shape) previously declined on polars — `unwind_polars` only accepted a scalar literal list. Now explodes a `List`-dtype column reference (`with_columns` copy → `filter(list.len() > 0)` → `explode`), matching the pandas oracle exactly: empty-list/null cells → 0 rows, nulls *within* a list survive, source column retained; the pre-filter also makes it independent of the polars-2.0 `empty_as_null` default. Non-list columns, name collisions, unknown identifiers, and nested-list literals still decline (honest NIE, no pandas bridge). Differential fuzz vs pandas over ~3500 collect→UNWIND→{RETURN, grouped, filtered, ORDER BY, aggregate, nested-unwind} queries: 0 disagreements. (The `MATCH ... WITH collect(..) UNWIND .. MATCH` form — IC6 with a trailing re-traversal — is compile-rewritten into WITH→MATCH reentry, a separate path.) Tests in `test_engine_polars_row_pipeline.py`.
- **GFQL polars engine natively runs the LDBC IC11/IC6 undirected variable-length bindings table (`-[*1..k]-`), unblocking the cross-alias `WHERE NOT(person=friend)` clause**: a bounded UNDIRECTED variable-length MATCH that materializes a bindings row table (`rows(binding_ops=...)`, emitted whenever a downstream clause — e.g. a cross-alias same-path `WHERE NOT a = b` / `a <> b` — references two node aliases on the path) previously declined on polars (`binding_rows_polars` hard-NIE'd every undirected multihop), while pandas handled it. This was the single residual blocking the official LDBC IC11/IC6 queries on polars (the cross-alias WHERE lowering itself was already native — directed var-length + `WHERE NOT a = b` already matched pandas). `binding_rows_polars` now supports undirected var-length with **`min_hops == 1`** via a doubled-pair join with immediate-backtrack avoidance, an exact port of the pandas oracle `_gfql_multihop_binding_rows` (`avoid_immediate_backtrack=True`): a `__prev__` marker (seeded null, dtype-matched to the id column) drops immediate backtracks each hop (Kleene mask: null prev kept), and the step-pair set reproduces pandas' edge multiplicity exactly — each non-loop edge contributes each directed orientation TWICE (so a length-1 pair appears x2, length-2 x4) while self-loops contribute `(u,u)` x2 only (not double-counted). The multiplicity rule was derived by instrumenting pandas' `step_pairs` (which flow from the var-length `edge_op.execute` hop + `orient_edges`), NOT by reading code alone. Scoped to `min_hops == 1` because that is where the raw-edge reconstruction provably matches pandas: every edge is trivially a length-1 path so the var-length hop's backward pruning removes nothing; `min_hops == 0` (zero-hop, undoubled) and `min_hops >= 2` (backward-pruned / long-walk divergence) still **decline with an honest `NotImplementedError`** rather than risk silent-wrong multiplicities. Differential fuzz vs the pandas oracle over ~2500 random graphs (self-loops, parallel + antiparallel edges, `*1..2`…`*1..5`, all WHERE/RETURN variants): 0 disagreements; the out-of-scope windows decline (never diverge). Tests in `test_engine_polars_binding_rows.py` (parity + multiplicity/backtrack/self-loop/string-id pins + `*0..2`/`*2..3`/`*2..2` decline pins).
- **GFQL polars engine natively runs multi-source (node-cartesian) MATCH (#1273)**: comma-separated disconnected aliases — `g.gfql("MATCH (a {..}), (b {..}) RETURN a.id, b.id")` — previously declined on polars (the `rows(binding_ops)` `node_cartesian` branch hard-NIE'd) while pandas handled them. New `_cartesian_node_bindings_polars` mirrors the pandas oracle `_gfql_cartesian_node_bindings_row_table`: each alias is independently filtered, projected into the per-alias lookup schema (bare `alias` id, `alias.id`, `alias.<prop>`, and pandas' leaked `alias.alias=True` flag incl. property-shadowing), then left-major cross-joined for row-order parity. Scalar/aliased/`count(*)` projections over the cartesian match pandas exactly (verified end-to-end); whole-entity `RETURN a, b` stays an honest NIE (separate projection surface). Two parity-safe declines mirror shapes where pandas itself errors (proven on master, so both engines fail identically, never diverge): an anonymous node op, and ≥4 named aliases (pandas' bare-id merge residue collides). Differential fuzz vs pandas over ~10k lowered cases: 0 disagreements. Tests in `test_engine_polars_binding_rows.py`.
- **GFQL polars chain — variable-length node aliases are hop-distance gated; the chain is now silent-wrong-free (#1741)**: a node named after a variable-length edge (`g.gfql("MATCH (a)-[*1..2]-(b) RETURN b")`) previously carried its alias regardless of hop distance, so an undirected walk that backtracked into the seed wrongly returned it (pandas correctly excludes it — trail semantics). The polars chain now auto-injects the #1741 hop-distance label (name resolved against the user's node columns via the `reserved_columns` registry, so a user column named like the internal one is never clobbered) and applies pandas' alias `[min_hop, max_hop]` window in `_apply_node_names`. The one shape the gate can't yet cover — a node alias after a **forward/reverse `min_hops>1`** edge, whose labels need pandas' layered backward walk (not ported) — now **declines with an honest `NotImplementedError` (#1748)** instead of returning nodes outside the window; the decline is precise (differential vs pandas: 30/30 named shapes NIE, 30/30 unnamed run and match, 0 over-decline). 4-engine A/B on the stacked build: pandas/cudf 144/144, polars/polars-gpu 112 agree / **0 disagree** / 32 honest-NIE — zero silent-wrong shapes remain on the polars chain. Adapts the mechanism from the retired #1742 (declined undirected varlen+alias, now natively gated). Tests: `TestVarlenAliasHopGate` (pandas-parity across directions, the `*2..3` decline + unnamed-still-runs pins, a column-collision test).
- **GFQL native polars `label_node_hops` on the plain BFS — correct, direction-dependent hop labels (#1741 groundwork)**: the polars eager hop now emits pandas-parity node hop-distance labels instead of declining `label_node_hops`. The labeling rule is DIRECTION-DEPENDENT (the subtle part, derived by differential fuzz vs the pandas oracle, not by reading pandas alone): forward/reverse label EVERY destination of a hop first-wins (matching pandas `hop.py` `new_node_ids`), so a seed re-entered at hop 1 IS labeled; undirected labels destinations MINUS everything already visited and pre-seeds the seen-set with the seeds, so a seed re-reached by a backtracking walk stays NULL (its shortest-path distance). Two correctness fixes over the first cut: (1) the undirected seed pre-seed must NOT depend on `return_as_wave_front` — a seed filtered out of the frontier by `source_node_match` or suppressed in wave-front mode was being re-labeled (A/B vs pandas over direction × hops × `return_as_wave_front` × `source_node_match` × `destination_node_match` × 10 graphs: was 456/24, now 480/480); (2) a requested `label_node_hops` name that collides with an existing column is redirected to `<name>_1` like pandas' `resolve_label_col`, not the polars left-join auto-suffix `<name>_right`/`DuplicateError`. `label_edge_hops` and `min_hops>1` labels stay honest NIEs. Amplified `test_engine_polars_hop.py` (`TestHopLabelsDifferential` over the seed-filter / wave-front / multi-seed axes, an explicit direction-asymmetry contrast pin, and a collision test).
Expand Down
104 changes: 83 additions & 21 deletions graphistry/compute/gfql/lazy/engine/polars/row_pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -1057,17 +1057,31 @@ def _names(lf: pl.LazyFrame) -> List[str]:
sem = EdgeSemantics.from_edge(op)
if sem.is_multihop:
# Bounded directed var-length (`-[*1..k]->`, graph-bench q3) is
# supported via iterative pair joins; everything else declines:
# unbounded (`[*]`, needs fixed-point + termination error),
# undirected multihop (immediate-backtrack avoidance not ported),
# and aliased var-length edges (pandas rejects those outright).
# supported via iterative pair joins. Bounded UNDIRECTED var-length
# with min_hops == 1 (`-[*1..k]-`, the LDBC IC11/IC6 shape) is now
# also supported via a doubled-pair join with immediate-backtrack
# avoidance (see the execution branch below). Everything else declines:
# unbounded (`[*]`, needs fixed-point + termination error), aliased
# var-length edges (pandas rejects those outright), and undirected
# var-length with min_hops != 1 (`-[*0..k]-` / `-[*2..k]-`): pandas'
# step_pairs come from the var-length `edge_op.execute` hop, whose
# backward hop-window pruning / zero-hop handling changes the edge
# multiplicity in a way this raw-edge reconstruction only reproduces
# for min_hops == 1 (every edge is trivially a length-1 path, so no
# pruning occurs) — fuzz-verified vs the pandas oracle. Decline the
# rest honestly rather than risk silent-wrong multiplicities.
if (
op.direction == "undirected"
or bool(op.to_fixed_point)
bool(op.to_fixed_point)
or (op.max_hops is None and op.hops is None)
or isinstance(op._name, str)
):
return None
if op.direction == "undirected":
_resolved_min = op.min_hops if op.min_hops is not None else (
op.hops if op.hops is not None else 1
)
if _resolved_min != 1:
return None
if op.direction not in ("forward", "reverse", "undirected"):
return None
if any(
Expand Down Expand Up @@ -1174,23 +1188,71 @@ def _names(lf: pl.LazyFrame) -> List[str]:
return None
min_hops = int(min_hops_value)
max_hops = int(max_hops_value)
pairs = oriented.select(["__from__", "__to__"])
state_cols = _names(state)
reachable = [state] if min_hops == 0 else []
current = state
# Lazy: build all max_hops iterations (no eager .height early-break —
# empty intermediates lazily join to empty, so the result is
# identical; the pandas break is an optimization, not semantics).
for _hop in range(1, max_hops + 1):
current = (
current.join(pairs, left_on="__current__", right_on="__from__", how="inner")
.drop("__current__")
.rename({"__to__": "__current__"})
.select(state_cols)
if sem.is_undirected:
# Bounded UNDIRECTED var-length, min_hops == 1 (gated above): the
# LDBC IC11/IC6 `-[*1..k]-` shape. Mirror the pandas oracle
# (`_gfql_multihop_binding_rows`, avoid_immediate_backtrack=True)
# EXACTLY, including its edge multiplicity: pandas' `step_pairs`
# come from the undirected var-length hop + `orient_edges`, which
# emits each NON-loop edge as (u,v)x2 AND (v,u)x2, and each
# SELF-loop as (u,u)x2 (loops are not double-counted). Reconstruct
# that here: `exec_rows` = both directions of non-loops + one row
# per self-loop; the final `pairs` doubles `exec_rows`
# (fuzz-verified vs pandas over random graphs incl. self-loops,
# parallel + antiparallel edges). A `__prev__` column (seeded null)
# carries the just-left node so each hop can drop immediate
# backtracks (`__to__ == __prev__`), matching pandas' Kleene mask
# (null prev -> kept).
normal = edges_f.filter(pl.col(src) != pl.col(dst))
loops = edges_f.filter(pl.col(src) == pl.col(dst))
fwd = normal.select([pl.col(src).alias("__from__"), pl.col(dst).alias("__to__")])
rev = normal.select([pl.col(dst).alias("__from__"), pl.col(src).alias("__to__")])
loop = loops.select([pl.col(src).alias("__from__"), pl.col(dst).alias("__to__")])
exec_rows = pl.concat([fwd, rev, loop], how="vertical")
pairs = pl.concat([exec_rows, exec_rows], how="vertical")
prev_col = "__prev__"
reachable = [state.select(state_cols)] if min_hops == 0 else []
# Seed the backtrack marker with the SAME dtype as __current__ so a
# non-Int64 node id (e.g. string ids) compares/concats cleanly.
current = state.with_columns(
pl.lit(None).cast(state.collect_schema()["__current__"]).alias(prev_col)
)
if _hop >= min_hops:
reachable.append(current)
state = pl.concat(reachable, how="vertical") if reachable else state.limit(0)
for _hop in range(1, max_hops + 1):
joined = current.join(
pairs, left_on="__current__", right_on="__from__", how="inner"
)
joined = joined.filter(
pl.col(prev_col).is_null() | (pl.col("__to__") != pl.col(prev_col))
)
# new prev = the node we are leaving (old __current__); new
# __current__ = __to__. Set prev BEFORE dropping __current__.
joined = (
joined.with_columns(pl.col("__current__").alias(prev_col))
.drop("__current__")
.rename({"__to__": "__current__"})
)
current = joined.select(state_cols + [prev_col])
if _hop >= min_hops:
reachable.append(current.select(state_cols))
state = pl.concat(reachable, how="vertical") if reachable else state.limit(0)
else:
pairs = oriented.select(["__from__", "__to__"])
reachable = [state] if min_hops == 0 else []
current = state
# Lazy: build all max_hops iterations (no eager .height early-break —
# empty intermediates lazily join to empty, so the result is
# identical; the pandas break is an optimization, not semantics).
for _hop in range(1, max_hops + 1):
current = (
current.join(pairs, left_on="__current__", right_on="__from__", how="inner")
.drop("__current__")
.rename({"__to__": "__current__"})
.select(state_cols)
)
if _hop >= min_hops:
reachable.append(current)
state = pl.concat(reachable, how="vertical") if reachable else state.limit(0)
else:
state = (
state.join(oriented, left_on="__current__", right_on="__from__", how="inner")
Expand Down
Loading
Loading