timely-util: add co_reduce2 and render SetDifference through it#37637
Closed
antiguru wants to merge 30 commits into
Closed
timely-util: add co_reduce2 and render SetDifference through it#37637antiguru wants to merge 30 commits into
antiguru wants to merge 30 commits into
Conversation
Single-time (one-shot SELECT) dataflows upgrade hierarchical Reduce and TopK plans to their monotonic variants. This used to run as a post-lowering pass, `refine_single_time_operator_selection`, that mutated the LIR node's plan enum in place after lowering had already computed the node's `AvailableCollections`. The in-place structural rewrite left the advertised collections describing the pre-upgrade variant, which blocks any later change that wants to re-advertise the node. Absorb the selection into lowering. `lower` reads `is_single_time()` from the dataflow and records it on the lowering `Context`, and the `TopK` and `Reduce` arms pick the monotonic variant at construction time via `as_monotonic(true)`. Lowering then computes `AvailableCollections` for the final variant. The flag is forced off while lowering the recursive bindings of a `LetRec`, whose values are not restricted to a single time, matching the old pass, which descended only into the `LetRec` body. Behavior is unchanged: `ReducePlan::keys()` and `extract_mfp_after` do not depend on the hierarchical sub-variant, and `TopK` advertises no arrangement, so the advertised collections are identical. `refine_single_time_consolidation` still relaxes `must_consolidate` after lowering. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Mirrors the arity field already present on MonotonicTopKPlan and BasicTopKPlan, so that render/lowering can later derive the plan's output arrangement layout from (group_key, arity). Pure plumbing: no behavior change.
Make `render_top1_monotonic` build its output as a group-key-keyed arrangement in the layout `permutation_for_arrangement` dictates, and have `render_topk` return that arrangement in the bundle for the MonotonicTop1 arm, mirroring `Reduce`. The winning row is thinned to the value columns; the group-key columns live in the key row. Because lowering does not yet advertise the arrangement, the bundle also carries a raw collection reconstructed from the arrangement via the permutation, so consumers planned with `input_key = None` (for example a Reduce over a DISTINCT ON) still find a collection. A later change advertises the arrangement and can drop the reconstruction. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Lowering now advertises the group-key arrangement that `render_top1_monotonic` builds, so a downstream consumer keyed on the same group key reuses it instead of forcing a redundant `ArrangeBy`. `MonotonicTopK`/`Basic` are unaffected: they key their arrangements by (hash, group_key), which a group-key-only consumer can't reuse. The render side keeps building both the arrangement and a reconstructed raw collection. Dropping the reconstruction (as originally intended) is unsound: `refine_single_time_operator_selection` can upgrade a `Basic` plan into `MonotonicTop1` after lowering has already computed `AvailableCollections` for the old (raw) variant, so a downstream consumer can still expect a raw collection directly. Unlike `Reduce::keys()`, which advertises the same arrangement for every `Hierarchical` sub-variant, `TopK`'s variants differ in what they arrange, so render can't tell which case it's in. Confirmed by a reproduced crash (`keys.raw <= collection.collection.is_some()` assertion) before adding the defensive raw reconstruction back. Adds a regression test proving the `ArrangeBy` is elided when two `DISTINCT ON` queries over the same monotonic source are joined on their group key. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…not both Task 3 defensively delivered a dual bundle (arrangement + reconstructed raw collection) because render couldn't tell whether a MonotonicTop1 plan was advertised as arranged by lowering, or was a Basic plan upgraded to MonotonicTop1 after lowering by TopKPlan::as_monotonic (whose advertisement stays raw). Add an explicit `arranged` flag to MonotonicTop1Plan, set true in the lowering-time construction and false in the post-lowering upgrade, and branch render on it to deliver exactly one shape.
`render_topk`'s `MonotonicTop1` arm computed `key` and `permutation` (via `permutation_for_arrangement`) unconditionally, but they are only used in the `!arranged` branch to reconstruct the raw collection from the arrangement. Move the computation into that branch; the `arranged` branch never used them and `key` duplicated work already done inside `render_top1_monotonic`. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The arranged=true correctness test spun up a bounded LOAD GENERATOR COUNTER source and waited with mz_sleep, which flakes. The arbitrary (non-prefix) group-key permutation reconstruction it exercised is already covered deterministically by the one-shot DISTINCT ON over a plain table (the arranged=false path, same permutation_for_arrangement output), and the arranged=true plan is covered by the join-elision EXPLAIN test. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…he variant Rebasing on MaterializeInc#37593 (which inlines single-time monotonic operator selection into lowering, running as_monotonic before the arrangement advertisement) means every MonotonicTop1 plan, including a one-shot SELECT DISTINCT ON upgraded during lowering, now advertises its group-key arrangement. That makes the `arranged` flag on MonotonicTop1Plan always true, so drop the field from the struct, create_from, as_monotonic, and the limit() destructure, and make render_topk's MonotonicTop1 arm unconditionally arrangement-only instead of branching on it. A consumer that needs the raw collection reconstructs it from the advertised arrangement via its permutation, same as for an index arrangement. Regenerated sqllogictest goldens affected by the now-always-arranged output: an explicit ArrangeBy on the group key is elided where the Top1 render used to require one, and Get/Reduce plans downstream of a MonotonicTop1 now see an arrangement instead of a raw collection. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Add a Feature Benchmark scenario, SetDifferenceCoArranged, that builds the set-difference anti-side both outer-join decorrelation and EXCEPT render: Threshold(Union(A, Negate(B))) over two inputs arranged on the difference key. When the inputs are co-arranged, rendering still inserts an intermediate ArrangeBy feeding the Threshold plus a UnionConsolidation over the concatenated inputs. WALLCLOCK and MEMORY_CLUSTERD then baseline the cost of those two operators. This is the measure-first artifact for the CLU-168 spike: a dedicated set-difference operator reading the two co-arranged traces directly would elide both, and re-running this scenario against that change measures the win. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Design doc for a dedicated SetDifference LIR operator that reads two co-arranged inputs directly and produces one output arrangement, eliminating the intermediate ArrangeBy (trace T, measured ~40% of anti-side arrangement memory) and UnionConsolidation from the Threshold(Union(Negate(A), B)) anti-side rendering. Phased: recognizer + node + no-op render first, fused binary operator second. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Fixes from adversarial spec review: - Operator semantics now group by key and sum diffs over values, making heterogeneous input thinning sound (outer_join l0 thinning=(#1) vs l1 ()). - Recognizer checks the full (key,permutation,thinning) triple, allows only a pure projection-to-key MFP, and declines on non-default temporal bucketing, non-Get/ArrangeBy arms, and non-projection MFPs. - Name the Phase 2 central risk: output-trace compaction against two upstream frontiers plus reduce-style per-key time-revisit. - Flag-gated refine pass called out as new plumbing; self-difference and heterogeneous-thinning test cases added; 40/51% marked as estimate. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Introduce the `SetDifference` LIR node and its `render_plan::Expr` counterpart. The node holds two heterogeneous children, `base` and `subtract`, plus an `ensure_arrangement` triple describing the output arrangement, mirroring `BasicThresholdPlan`. Wire it through `children`/`children_mut`, `depends_on_into`, the `render_plan` flattening and `Display`, the EXPLAIN verbose and default text arms, and the abstract-interpreter framework (`Interpreter` trait, both fold implementations, and the `SingleTimeMonotonic` analysis). Render it in phase 1 by reconstructing the dataflow it replaces: consolidate `Union(base, Negate(subtract))`, arrange by the ensure key via `ensure_collections`, and threshold through the existing basic path. Output is byte-identical to the `Threshold` this node stands in for. The node and render arm land in one commit because the `Expr` variant makes the `render.rs` match non-exhaustive, so the render arm must exist for the workspace to compile. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Add a flag-gated LIR-layer refine pass that rewrites the co-arranged anti-side shape Threshold::Basic(ArrangeBy(Union(base, Negate(subtract)))) into a single SetDifference node. Declines unless both arms are arranged on the threshold key via a projection-only read MFP, exactly one Union arm is a Negate, and neither arm carries a non-Direct temporal bucketing strategy (the render does not reproduce bucketing). Gated behind enable_fused_set_difference, dormant off. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Proves the fused set-difference recognizer (CLU-168) is behavior-preserving: EXCEPT and LEFT JOIN anti-side queries return identical rows with enable_fused_set_difference on vs off.
Spike resolution for the binary_frontier operator: output-trace compaction pinned to P = meet(upper_base, upper_subtract) for both inputs and the output, reduce-style per-key time-revisit over the union of dirty keys with synthetic times, forking reduce_trace's output machinery and mz_join_core's two-input skeleton. Feasible, medium complexity; differential-oracle test mandated. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…erence recognizer `arm_arrangement_matches` stripped a projection-only `Mfp` between a `Union` arm and its underlying arrangement, then compared the threshold key against the underlying arrangement key without remapping through the projection. The threshold key lives in the arm's output (post-projection) column space while the arrangement key lives in the arm's underlying (pre-projection) space, so a reordering or non-trivial projection produced a false `#i`-vs-`#i` match on physically distinct columns. Map the threshold key columns back through the projection (`output_col[i] = underlying_col[proj[i]]`) into underlying-column space and require exact ordered equality with the underlying arrangement key. Identity and no-projection arms are unchanged; reordering or shifting projections now decline, since the operator only serves an input keyed exactly as the output needs. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Replace the Phase-1 SetDifference render arm, which reconstructed Threshold(ArrangeBy(Union(base, Negate(subtract)))), with a fused binary_frontier operator that reads the two co-arranged inputs directly and produces one output arrangement, eliminating the intermediate union arrangement (trace T). The operator forks the two-input consumption skeleton of mz_join_core and the output-trace machinery of differential's reduce_trace, and reimplements the per-key signed-net walk (the ()-valued collapse of HistoryReplayer) with synthetic interesting times. Each input is read at its own frontier (cursor_through(upper_i)); the combined frontier P = meet governs only output sealing and compaction. Reading both inputs at P (as the spike proposed) straddles batches and panics. Add a maintained-view incremental correctness section to outer_join.slt covering retractions, key reappearance, net-zero keys, and net>1 multiplicity; results are identical Phase 1 vs Phase 2 (oracle), goldens unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ide LetRec bin/lint rejects std #[test]; switch the remap_*/arm_arrangement unit tests to #[mz_ore::test], matching the convention used elsewhere in this crate. Also stop refine_set_difference from descending into LetRec subtrees. The fused SetDifference operator's partial-order/synthetic-time path is untested under iterative re-evaluation, so declining to rewrite anywhere inside a recursive CTE (both its bindings and its body) is the safe default until that path is verified. Add a golden covering a WITH MUTUALLY RECURSIVE query with a self-referencing EXCEPT plus a trailing EXCEPT in the body, showing neither fuses while the existing non-recursive EXCEPT still does.
Extend the fused SetDifference recognizer so it captures the memory win for the outer-join anti-side, not only EXCEPT. An outer-join anti-side arm reads its arrangement with a projection (`Get::Arrangement l0 project=(#0)`), which renders to a raw collection and forces the fused operator to re-arrange it, reintroducing the very arrangement the fusion removes. The recognizer now strips that projection and records the key each arm's underlying arrangement advertises, in two new `LirRelationNode::SetDifference` fields `base_key` and `subtract_key`. The stripped arm exposes the existing arrangement via `Get::PassArrangements`, so the operator reads it directly. Stripping is sound because the operator sums diffs per key and ignores value columns, and the underlying arrangement's key datums equal the output key datums. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Look up `base` by `base_key` and `subtract` by `subtract_key` instead of the shared output key. For a projecting outer-join arm the recognizer now strips the projection and exposes the underlying arrangement, so the lookup hits and no per-arm re-arrangement is built. The key datums align across arms, so per-key net accounting co-iterates unchanged, and the output stays keyed by the output arrangement key. The `None` fallback re-arrange remains for arms that genuinely have no arrangement. Rewrites the outer_join.slt EXPLAIN goldens: the LEFT JOIN anti-side arm now reads `Get::PassArrangements l0` directly rather than a re-arranged `Get::Arrangement l0 project=(#0)`. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…tion An ArrangeBy anti-side arm can advertise its arrangement either through its own `forms` (keyed in output column space) or through the `input_key` of the input arrangement it reads (keyed in input column space). The recognizer compared `input_key` directly against the projection-remapped output key, which only aligns when `input_mfp` neither reorders nor drops columns. Reading such an arm directly by that key (now that a projecting wrapper is stripped instead of forcing a re-arrange) would collate different logical columns across the two arms under a non-identity `input_mfp`, producing a wrong difference. Require an identity `input_mfp` projection before matching on `input_key`. No plan in the corpus exercises this shape, so goldens are unchanged; the guard is defensive. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The per-key push loop drained each capability's updates into the batch builder without sorting. emit_deltas produces updates time-major, value-minor, but OrdValBuilder::push assumes input sorted by (key, val, time) and only compares each item against the last pushed. A key emitting multiple distinct values across multiple times therefore wrote a value-unsorted vals array, violating the batch's sorted-unique cursor invariant, corrupting value seeks and spine merges. Stable-sort each key's updates by value before draining, matching differential's own reduce. Add co_reduce_multi_value_per_key, which reads the output arrangement's batch stream and asserts stored values are ascending (the invariant the bug breaks; a summed readback cannot observe it). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Replace the variadic N-input co_reduce with a 2-input co_reduce2 built on a binary_frontier shell, matching set_difference_core. The two inputs may carry distinct trace types (key-equated, diff-pinned), which handles the Local-vs-Trace arrangement-flavor demux by monomorphization and dissolves the homogeneous Vec<Arranged<Tr>> type-unification blocker. Generalize Bu::Input to Default + PushInto<((K,V2),T,R2)> so RowRow output builders (whose input is a columnar stack, not Vec) are admissible. The per-key value-aware core is unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Replace the private set_difference_core with a call to the generic co_reduce2 over two co-arranged inputs. The per-key threshold arithmetic moves into a free set_difference_threshold closure, and the four Local/Trace flavor combinations demultiplex onto co_reduce2 by monomorphization. The trace-walk helpers now live in mz_timely_util::co_reduce. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Bound the number of dirty keys co_reduce2 processes per activation so it yields the timely worker instead of draining an unbounded dirty set in one shot. A round in progress is finalized atomically: the operator accumulates processed keys' updates into per-capability builders and only builds, ships, and seals once the round's dirty set fully drains. Until then it holds the round's output capabilities, leaves processed unadvanced, defers compaction, and re-activates itself. This preserves the invariant that a round is never sealed while a dirty key still owes output below upper_limit. The four set_difference call sites pass a fixed fuel of 1_000_000, with a TODO to track the differential/default_exert_logic config. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Existing tests exercise retraction only with unbounded fuel (co_reduce_incremental) and small fuel only with insertion (co_reduce_fuels). Add co_reduce_fuels_retraction to close that gap: a round that retracts many keys back to net-zero must fully drain and retract them even when the dirty set is fuel-split across many activations. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ing) Update the design doc to describe the shipped 2-input co_reduce2 (distinct trace types, fuel parameter, accumulate-then-ship-on-drain fueling) instead of the originally-scoped variadic homogeneous co_reduce, keeping the N-input generalization as noted future work. Drop the dangling set_difference_core reference and the cross-crate mz_join_core mention from co_reduce.rs's module doc. Rename co_reduce2's base/subtract parameters to input0/input1 so the generic operator doesn't carry set-difference vocabulary, note the O(N^2/fuel) cost of re-acquiring cursors per activation, and clarify that the set-difference fuel budget is a distinct knob from the exert/compaction-effort config. Co-Authored-By: Claude Sonnet 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Created in error. This work is included in #37595. Closing.