compute: fused SetDifference operator over co-arranged inputs (CLU-168)

[antiguru]

Draft. CLU-168. Stacks on #37592 (those commits drop out once it merges).

A dedicated SetDifference LIR operator for the co-arranged set-difference anti-side that outer-join decorrelation and EXCEPT render, replacing Threshold(ArrangeBy(Union(Negate(A), B))). It reads the two input arrangements directly and produces one output arrangement, eliminating the intermediate ArrangeBy (trace T). Behind enable_fused_set_difference, default off. Design: doc/developer/design/20260713_fused_set_difference.md.

Measured win (maintained EXCEPT, a=1M / b=500K, local mz_arrangement_sizes): the dataflow's arrangement memory drops from 64.2 MB / 6.0M records to 15.0 MB / 3.5M records. Trace T (25.8 MB) is eliminated and the diff step collapses from two arrangements to one; the input DistinctBy arrangements now dominate.

Structure:

• enable_fused_set_difference feature flag, default off in production.
• LirRelationNode::SetDifference { base, subtract, ensure_arrangement } with EXPLAIN, abstract-interpreter (monotonicity identical to Threshold), and keys() plumbing.
• A flag-gated LIR-layer recognizer that rewrites the co-arranged anti-side shape; declines on non-co-arranged arms, non-projection MFPs, reordering projections (key remapped through the projection), non-default temporal bucketing, non-Get/ArrangeBy arms, and inside LetRec.
• The fused operator (src/compute/src/render/set_difference.rs) reads the two input arrangements directly and produces one output arrangement. Output finalizes at P = meet(upper_base, upper_subtract); each input is read at its own frontier (reads past P are inert); input logical compaction pinned to P, physical to each input's frontier. Per-key two-source net with reduce-style interesting-times and synthetic-time closure.

Correctness: results are byte-identical to the Threshold(Union(..)) path, verified by result-equivalence and multi-round incremental (retraction / reappearance / net-to-zero / multiplicity) sqllogictest against the old path as oracle. EXPLAIN goldens for the recognizer, plus a recursive-CTE golden confirming the operator declines inside LetRec.

---

Update: extracted into co_reduce2 (later commits on this branch).

The fused operator body described above is extracted into a generic two-input reduce, co_reduce2, in mz-timely-util, and SetDifference now renders through it. co_reduce2 reduces two co-arranged inputs per key via a caller closure and produces one output arrangement. The two inputs may carry distinct trace types, unified only on the key GAT, owned key and value, and diff, so the Local/Trace arrangement flavors are handled by monomorphization rather than a hand-written operator. The threshold becomes the closure net = sum(input0) - sum(input1), removing ~585 lines from set_difference.rs. It is Row-agnostic and value-aware, living beside differential's own reduce. outer_join.slt stays byte-identical.

The operator gains a fuel budget that bounds the keys processed per activation and reschedules, so a large dirty set no longer monopolizes the worker. The output frontier is held until a round's dirty set fully drains, so an incomplete round is never finalized.

The projecting LEFT-JOIN anti-branch passthrough (per-arm base_key/subtract_key) is implemented, so the outer-join case realizes the win, not only EXCEPT.

Design: doc/developer/design/20260714_co_reduce_operator.md. A variadic-N co_reduce over homogeneous inputs remains deferred future work. New co_reduce2 unit tests cover value-aware output ordering, incremental retraction, fuel yielding, and retraction under a fuel-split round.

Remaining follow-up (not a blocker): per-key recomputation is O(n^2) in per-key history; the Feature Benchmark scenario SetDifferenceCoArranged (included) should be run in CI as the A/B.

🤖 Generated with Claude Code