Skip to content

compute: park future updates in a merge batcher wrapper - #39928

Draft
antiguru wants to merge 11 commits into
MaterializeInc:mainfrom
antiguru:temporal-batcher-wrapper
Draft

antiguru wants to merge 11 commits into
MaterializeInc:mainfrom
antiguru:temporal-batcher-wrapper

Conversation

@antiguru

Copy link
Copy Markdown
Member

Spike of a leaner alternative to #37589: temporal bucketing as a thin wrapper around the existing merge batchers instead of a rewritten batcher. TemporalBatcher routes updates at or beyond max(frontier edge, as_of) + summary into the existing time-major bucket chain when they are pushed, peels matured buckets into the inner batcher at each seal, and reports the meet of both frontiers. It works over all three batcher arms through a small TimedUpdates trait, and over iterative timestamps by bucketing on the outer coordinate against frontier elements whose inner time is the minimum.

The wrapper replaces the temporal_bucket operator at every rendering site except the accumulable reduce, whose tuple diff the bucket storage cannot yet hold. The lowering plumbing and the operator stay in place for now. The dataflow as_of reaches batchers through a scoped thread-local set while a dataflow renders, and the summary through a process-global installed from the worker config.

Adds model tests and proptests against the plain batcher (total and Product times, spilling, stepped hydration uppers) and a criterion bench in mz-timely-util. This draft exists to run the feature benchmark against the operator. A local comparison showed steady-state parity and slower hydration, because the future tail passes key formation and the arrange chunker before it parks.

🤖 Generated with Claude Code

antiguru and others added 11 commits October 17, 2026 16:57
Move the temporal bucket operator's spillable bucket storage into
`mz_timely_util::temporal::bucket` as `TimeMajorBucket`, so a batcher
in timely-util can reuse it. The compute-only `MzData` bounds become
the columnar bounds the type needs, the tests use `u64` times, and
`BucketTimestamp for u64` leaves the test module. The operator uses the
moved type unchanged.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
`TemporalBatcher` wraps any differential `Batcher` and keeps updates at
least θ past the last sealed frontier in the existing bucket chain of
time-major buckets, so the inner batcher does not re-merge them on every
seal. Seals peel the chain at the frontier's inner-minimum key and feed
the peeled updates through a chunker into the inner batcher, which
decides readiness. The first seal rebuilds the inner batcher, because
the snapshot and its future tail arrive before the batcher learns the
frontier.

Buckets are keyed by a new `BucketTime` trait, the identity for totally
ordered times, and track a lower bound of their times for the wrapper's
frontier. The wrapper reads output chunks through `TimedUpdates`,
implemented for the three output types of the arrange batcher arms.

Adds model tests against the plain batcher (hand-written, hydration
shaped, spilled, and proptest) and a criterion bench.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
At the ArrangeBy ok arrangement, wrap all three batcher arms in
`TemporalBatcher` when the timestamp is `mz_repr::Timestamp`, and stop
applying the temporal bucketing operator in `ensure_collections`.
Iterative timestamps keep the plain batchers. `apply_worker_config`
installs θ from `compute_temporal_bucketing_summary`.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Implement `BucketTime` for `Product` by its outer coordinate, and use
the batcher wrapper for iterative timestamps at the ArrangeBy ok
arrangement too. A seal parks updates only relative to the least key
whose inner-minimal time is in the frontier, so updates that might be
ready always sit in the inner batcher. Adds model tests over
`Product<u64, u64>` times.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
The temporal batcher marked itself rebuilt at its first seal even when
that seal's upper had no inner-minimum element, so a later seal with a
far bound would leave the hydration tail in the inner batcher. Rebuild
at the first seal that has a far bound instead. With advancing uppers
the case cannot arise, since an inner-minimum element only follows
another, so the change is defensive. Rename the stats to match.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
…ites

Replace the temporal bucketing operator with batchers that park future
updates at Reduce input arrangements and the monotonic consolidation,
TopK arrangements and consolidations, linear join stage arrangements,
the iterative index export, and `columnar_consolidate`, which covers
consolidating Unions and `LetRec` bindings. `RenderTimestamp` now
requires `BucketTime` keyed by `mz_repr::Timestamp`, so the wrapped
batchers need no per-site dispatch.

The accumulable Reduce plan keeps the operator: its arrangement carries
`(Vec<Accum>, Diff)` diffs, which the bucket storage cannot consolidate
because the diff does not add a columnar ref. The lowering's
`TemporalBucketing` strategies stay plumbed but mostly go unused.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Match `compute_temporal_bucketing_summary` and timely's `Summary`
naming in fields, setters, the process-global, comments, tests, and the
bench.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
The batcher inferred hydration from its first seal and rebuilt its inner
batcher there, moving the snapshot's future tail into the bucket chain.
The dataflow builder now publishes the dataflow `as_of` in a scoped
thread-local that `TemporalBatcher::new` reads. With it, the batcher
classifies pushes before its first seal against `as_of` plus the
summary, so the tail parks on arrival and the snapshot stays in the
inner batcher, and it joins each seal's edge with `as_of` as the
bucketing operator does. The chain is shaped around `as_of` up front,
so the tail does not land in one bucket spanning the domain. The
rebuild and its stats go away. Without an `as_of` nothing parks before
the first seal.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Every site that bucketed a columnar edge now parks future updates in its
batcher, so `MaybeBucketByTime::maybe_apply_temporal_bucketing` has no
callers. The `Vec` variant stays for the accumulable reduce path.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant