Repository navigation
Conversation
antiguru
force-pushed
the
temporal-bucketing-batcher
branch
from
May 7, 2026 06:48
fedf72d to
fe142dc
Compare
Copy `MergeBatcher`, `Merger`, `InternalMerge`, `InternalMerger`, and `VecMerger` from `differential_dataflow::trace::implementations::merge_batcher` into `mz_timely_util::merge_batcher`. The next commit layers temporal bucketing on top of this vendored skeleton; vendoring it gives us control over the chain representation without forking differential. Verbatim port apart from minor module wiring; no behavior change. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Replaces the source-side temporal-bucketing operator on the arrangement path with a bucket-aware merge batcher that integrates directly into mz_arrange call sites. Layered on top of the vendored `mz_timely_util::merge_batcher`, the new `TemporalBucketingMergeBatcher` keeps current data on a flat `chains: Vec<Vec<M::Chunk>>` exactly like plain `MergeBatcher` and routes only the future-stamped `kept` side of each seal extract into a `BucketChain` of `MergeBucket`s. Subsequent seals peel mature buckets out of the bucket chain and merge their contents with the current flat extract. The seal contract is preserved: only updates strictly less than `upper` are emitted. `Batcher::frontier` reports the actual minimum held timestamp by merging the lowest non-empty bucket once per seal and consulting `Merger::extract`. Subsequent seals see a single chain in that bucket, so the work is amortised. `CollectionBundle::ensure_collections` reads `ENABLE_TEMPORAL_BUCKETING` and dispatches the row-row arrangement build through a new per- `RenderTimestamp` trait `MaybeTemporalRowRowArrange`. The trait is a sibling to `RenderTimestamp`, not a supertrait, so the bucketing precondition does not leak into the generic timestamp interface. Iterative `Product<...>` always uses the non-temporal batcher. Several seal-time fast paths keep per-push and per-seal cost close to plain `MergeBatcher` when the bucket chain is empty: * Held-records counter short-circuits `seal` when there is no future data and the current extract produced no `kept` side, skipping the full peel/merge/restore/recompute_frontier cycle. * `Merger::time_range` reports per-chunk `(lower, upper)` antichains so `MergeBucket::split` can short-circuit when data lies entirely on one side of the split timestamp. * `kept` is routed into the bucket chain as a single chain rather than per chunk, so the destination bucket adopts the consolidated chain instead of pairwise-merging it back into existence. * The slow-path route walks the chain via the lower frontier returned by `extract`, jumping directly to the next non-empty bucket instead of iterating every boundary. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
The arrange-side `TemporalBucketingMergeBatcher` (per-`ArrangeBy`) fully covers the source-side delay this operator used to provide. Keeping both pays a double bucketing cost on every dataflow that touches the source. Remove the operator and its dyncfg `TEMPORAL_BUCKETING_SUMMARY`; the bucket chain advances naturally with the seal upper. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
Cherry-picked from MaterializeInc#35935. Two scenarios exercise the bucketing path: * TemporalFilterIndexed: index hydration over a temporal-filtered view of a 10M-row materialized view. * TemporalFilterSustainedInsert: per-insert latency through a pre-warmed indexed temporal view.
antiguru
force-pushed
the
temporal-bucketing-batcher
branch
from
May 7, 2026 12:59
dceab0c to
3490576
Compare
… start Replace the per-seal merge+extract over the lowest non-empty bucket with a walk that returns that bucket's start. The bucket start is a valid lower bound on every held time in the bucket, so the contract `arrange_core` relies on (`Batcher::frontier()` returning a lower bound on held data) is preserved. The reported frontier may lag the true minimum by up to one bucket span, which delays capability release on the arrangement output but does not affect correctness. `BucketChain::iter_mut` had a single caller (the old `recompute_frontier`) and is replaced by a non-mut `iter`, since the frontier walk now reads but does not modify buckets. Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>
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
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.
Motivation
Temporal bucketing currently runs as a separate Timely operator (
extensions::temporal_bucket) sitting upstream of everymz_arrangecall.The operator buffers future-stamped updates in a
BucketChain<MergeBatcherWrapper>and re-emits them once the input frontier crosses bucket boundaries, which means data passes through two separate batcher hierarchies before reaching the arrangement spine.Pushing the bucketing into the arrangement's own
Batcherremoves the operator hop, lets the spine see fewer bucket-aligned batches, and avoids the timestamp rewrites that the operator performed to align releases.Description
mz_timely_util::merge_batchervendorsMergeBatcher,Merger,InternalMerge,InternalMerger, andVecMergerfrom differential-dataflow.Vendoring is required because the bucket-aware logic operates on a merger's chunks, frontier, and stash --- machinery that is not exposed through the public
Batchertrait alone.The vendored skeleton is byte-equivalent to upstream and lives in its own commit; the temporal
MergeBucketandTemporalBucketingMergeBatcherare added on top in a separate commit.TemporalBucketingMergeBatcher<I, C, M>is a hybrid: it has both a flatchains: Vec<Vec<M::Chunk>>field and abucket_chain: BucketChain<MergeBucket<M>>.Every input chunk lands in the flat chains via
insert_chainwith the same geometric merging that plainMergeBatcherperforms, so the input path does not touch the bucket chain at all.At seal time the flat chains are merged into a single sorted chain,
Merger::extractis run against the sealupper, and the result is split intoflat_ready(t < upper) andkept(t >= upper).The
keptchain --- and only that chain --- is routed into the bucket chain as a single geometric level so the destination bucket adopts it without pairwise re-merging.The bucket chain is then peeled at the same
upper; mature buckets release their data, the resulting chunks are merged into a single sortedpeeled_chain, andpeeled_chainis merged withflat_readyto produce the final readied chunks the builder consumes.Peel's contract guarantees the peeled buckets are entirely below
upper, so no second extract is needed againstupperafter merging.This layout keeps the bucket chain empty for workloads whose data is at or below the current frontier (the common case):
keptis empty at every seal, the bucket chain is never seeded, and per-push and per-seal cost reduces to plainMergeBatcherwith no overhead.The bucket chain only pays its splitting and merging overhead for the genuine future tail of the input.
Bucket-chain shape is restored with a bounded fuel budget at the end of seal;
push_containerdoes no bucket-chain maintenance.Several seal-time fast paths keep per-push and per-seal cost close to plain
MergeBatcherwhen the bucket chain is empty:sealwhen there is no future data and the current extract produced nokeptside, skipping the full peel/merge/restore/recompute_frontier cycle.Merger::time_rangereports per-chunk(lower, upper)antichains soMergeBucket::splitcan short-circuit when data lies entirely on one side of the split timestamp.extract, jumping directly to the next non-empty bucket instead of iterating every boundary.BucketChain::restorereturns immediately when every bucket already satisfies the chain property.MergeBucket<M>holdsVec<Vec<M::Chunk>>(a geometric chain of sorted chunks, mirroringMergeBatcher's flat chain invariant) and implementstemporal::Bucketso the chain can split it on demand.MergeBucket::split(timestamp, fuel)merges its chains into a single sorted chain and runsMerger::extract(merged, [timestamp], ...)to partition the data into a lower bucket (t < timestamp) and an upper bucket (t >= timestamp).Fuel is decremented by the record count touched so the chain's amortised splitting accounting still works.
Batcher::frontier()reports the lower bound of the lowest non-empty bucket --- the same temporal granularity the rest of the bucketing operates at.A bucket's start is a valid lower bound on the times in that bucket, so the contract that
arrange_corerelies on is preserved.The reported frontier can lag the true minimum by up to one bucket span, which delays capability release on the arrangement output but does not affect correctness.
After seal, all held data lives in
bucket_chain, so the frontier walk only inspects that side.mz_timely_util::columnarexposesCol2ValTemporalBatcher<K, V, T, R>--- the temporal counterpart ofCol2ValBatcher--- by wrappingTemporalBucketingMergeBatcheraround the sameChunkerandColInternalMergerused elsewhere.CollectionBundle::ensure_collectionsreadsENABLE_TEMPORAL_BUCKETINGfrom the dataflow'sConfigSetand threads the flag down to the row-row arrangement build via the newMaybeTemporalRowRowArrangetrait insrc/compute/src/extensions/arrange.rs.The trait has one impl per concrete
RenderTimestamp:mz_repr::TimestamppicksCol2ValTemporalBatcherwhen the dyncfg is on andCol2ValBatcherotherwise;Product<mz_repr::Timestamp, PointStamp<u64>>(iterative scopes) always picks the plain batcher because the timestamp is partially ordered and not bucketable.The trait is intentionally a sibling of
RenderTimestamprather than a supertrait, so the bucketing precondition (T: BucketTimestamp) does not leak into the generic timestamp interface; the bound is added explicitly on the impl block that hostsarrange_collection.top_k.rs,reduce.rs,threshold.rs,typedefs.rs,columnation.rs, andupsert_continual_feedback_v2.rsflip their imports fromdifferential_dataflow::trace::implementations::merge_batcherto the vendoredmz_timely_util::merge_batcherso that the codebase has a single source for these types.The source-side
extensions::temporal_bucketoperator andTEMPORAL_BUCKETING_SUMMARYdyncfg are removed: the arrange-side batcher fully covers the delay they provided, and keeping both pays a double bucketing cost on every dataflow that touches the source.