Skip to content

compute: temporal-bucketing batcher - #36427

Closed
antiguru wants to merge 5 commits into
MaterializeInc:mainfrom
antiguru:temporal-bucketing-batcher
Closed

antiguru wants to merge 5 commits into
MaterializeInc:mainfrom
antiguru:temporal-bucketing-batcher

Conversation

@antiguru

@antiguru antiguru commented May 6, 2026 •

Copy link
Copy Markdown
Member

Motivation

Temporal bucketing currently runs as a separate Timely operator (extensions::temporal_bucket) sitting upstream of every mz_arrange call.
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 Batcher removes 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_batcher vendors MergeBatcher, Merger, InternalMerge, InternalMerger, and VecMerger from 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 Batcher trait alone.
The vendored skeleton is byte-equivalent to upstream and lives in its own commit; the temporal MergeBucket and TemporalBucketingMergeBatcher are added on top in a separate commit.

TemporalBucketingMergeBatcher<I, C, M> is a hybrid: it has both a flat chains: Vec<Vec<M::Chunk>> field and a bucket_chain: BucketChain<MergeBucket<M>>.
Every input chunk lands in the flat chains via insert_chain with the same geometric merging that plain MergeBatcher performs, 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::extract is run against the seal upper, and the result is split into flat_ready (t < upper) and kept (t >= upper).
The kept chain --- 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 sorted peeled_chain, and peeled_chain is merged with flat_ready to 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 against upper after merging.

This layout keeps the bucket chain empty for workloads whose data is at or below the current frontier (the common case): kept is empty at every seal, the bucket chain is never seeded, and per-push and per-seal cost reduces to plain MergeBatcher with 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_container does no bucket-chain maintenance.

Several seal-time fast paths keep per-push and per-seal cost close to plain MergeBatcher when the bucket chain is empty:

  • A 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.
  • 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.
  • BucketChain::restore returns immediately when every bucket already satisfies the chain property.

MergeBucket<M> holds Vec<Vec<M::Chunk>> (a geometric chain of sorted chunks, mirroring MergeBatcher's flat chain invariant) and implements temporal::Bucket so the chain can split it on demand.
MergeBucket::split(timestamp, fuel) merges its chains into a single sorted chain and runs Merger::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_core relies 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::columnar exposes Col2ValTemporalBatcher<K, V, T, R> --- the temporal counterpart of Col2ValBatcher --- by wrapping TemporalBucketingMergeBatcher around the same Chunker and ColInternalMerger used elsewhere.
CollectionBundle::ensure_collections reads ENABLE_TEMPORAL_BUCKETING from the dataflow's ConfigSet and threads the flag down to the row-row arrangement build via the new MaybeTemporalRowRowArrange trait in src/compute/src/extensions/arrange.rs.
The trait has one impl per concrete RenderTimestamp: mz_repr::Timestamp picks Col2ValTemporalBatcher when the dyncfg is on and Col2ValBatcher otherwise; 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 RenderTimestamp rather 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 hosts arrange_collection.

top_k.rs, reduce.rs, threshold.rs, typedefs.rs, columnation.rs, and upsert_continual_feedback_v2.rs flip their imports from differential_dataflow::trace::implementations::merge_batcher to the vendored mz_timely_util::merge_batcher so that the codebase has a single source for these types.

The source-side extensions::temporal_bucket operator and TEMPORAL_BUCKETING_SUMMARY dyncfg 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.

@antiguru
antiguru force-pushed the temporal-bucketing-batcher branch from fedf72d to fe142dc Compare May 7, 2026 06:48
antiguru and others added 4 commits May 7, 2026 14:35
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
antiguru force-pushed the temporal-bucketing-batcher branch from dceab0c to 3490576 Compare May 7, 2026 12:59
… 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]>
@antiguru antiguru closed this Oct 4, 2026
@antiguru
antiguru deleted the temporal-bucketing-batcher branch November 12, 2026 10:09
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