Repository navigation
Conversation
Design document for the temporal-bucketing merge batcher: motivation (replace the lowering's static may-analysis with a free runtime is-check in the batcher), the batcher design (three-way seal split, near/far threshold, attained bucket bounds, frontier contract with the arrange operator), uniform coverage wiring, performance across the three regimes, correctness and testing, and the rollout plan. Co-Authored-By: Claude Fable 5 <[email protected]>
Add TemporalBucketingMergeBatcher, a drop-in differential Batcher that avoids the plain merge batcher's quadratic behavior on future-stamped updates (each held update is re-merged at every seal until the frontier passes it). Each seal partitions the held updates three ways against the upper and upper + threshold: ready updates are emitted and near-future updates return to the flat chains, both exactly like the plain batcher, while far-future updates are parked in a BucketChain and only touched again when splits or peels require it, amortized logarithmic in their distance from the frontier. Without far-future data, the seal is the plain batcher's code plus one branch. Buckets carry the attained (min, max) times of their contents, established during the extract walks that move the data anyway. Bounds let splits whose midpoint misses a bucket's data move the bucket wholesale, and a bucket whose contents consolidate away drops its content, since stale bounds would feed the reported frontier times for which no data exists. That frontier must never fall below a sealed upper (the arrange operator turns it into capabilities whose times become batch hints downstream), and a soft panic in seal guards this. Observability is a cumulative records-touched counter, a per-seal high-water mark, and a debug log line for seals that did bucket work. The batcher is generic over TemporalMerger, its chain-merging engine: differential's Merger role plus a three-way bounds-tracking extract. PagedColumnMerger implements it over the ColumnMergeBatcher family's paged column chains, so held data, in particular the rarely-touched far-future tail, can spill under memory pressure. PagedColumn gains a record count in its pageout metadata so the batcher can account held records without materializing chunks. Tests: a model-based proptest (each seal must emit exactly the not-yet-emitted updates below the upper, consolidated, with an exact frontier) and a proptest simulating the arrange operator's capability protocol against a downstream trace import, with updates that can cancel, both with an always-resident and a page-everything pager. Co-Authored-By: Claude Fable 5 <[email protected]>
Add ColTemporalMerger, the TemporalMerger over the columnation-based ColInternalMerger: a delegation wrapper plus that engine's three-way bounds-tracking extract walk. The paged engine is the primary target, but this adapter is what makes the temporal batcher deployable now, for three reasons. The columnation engine is the default everywhere, so the feature could otherwise not roll out independently of the paged batcher's own rollout. Paged builders exist only for the row-row and val-row arrangement families, so covering every batcher site uniformly is only expressible on the columnation engine. And CI runs the columnation engine nearly exclusively, so temporal code paths would otherwise go unexercised there. The adapter is deleted together with the engine. The model and arrange-protocol proptests gain columnation instantiations, behavioral unit tests cover the seal classification, bucket release, cross-seal consolidation, and hydration cost bounds, and a criterion benchmark compares against the plain batcher: parity on no-future and raced-ahead workloads is a hard requirement and holds at 0.98-0.99, while far-future workloads improve by an order of magnitude. Co-Authored-By: Claude Fable 5 <[email protected]>
ggevay
force-pushed
the
temporal-bucketing-merge-batcher
branch
6 times, most recently
from
July 12, 2026 15:48
c897a69 to
1a3c8f8
Compare
Add the enable_compute_temporal_bucketing_batcher dyncfg (default off). When set, every merge batcher in dataflow rendering on a totally ordered timestamp becomes a temporal-bucketing batcher and the standalone temporal_bucket operator is suppressed: batchers park far-future updates based on the data they actually observe, regardless of the lowering's ArrangementStrategy. There is deliberately no reasoning about which sites "need" bucketing: the batcher's fast path makes unnecessary coverage free, so uniform coverage replaces the per-site analysis. The flag takes precedence over enable_compute_temporal_bucketing (the operator_bucketing_summary helper holds that rule in one place), is captured once per dataflow in the rendering Context, and composes with enable_column_paged_batcher: with both set, the affected arrangements use the temporal batcher over paged chains. Batcher selection dispatches per render timestamp through a new MaybeTemporalArrange trait (mirroring MaybeBucketByTime): the default method bodies ignore the flag, which is correct for the iterative-scope Product timestamp since partially ordered timestamps are not bucketable, and mz_repr::Timestamp overrides them with the temporal dispatch. Covered sites: the ArrangeBy ok and err arrangements, Reduce input arrangements and the monotonic consolidation, TopK, Threshold, the linear-join stages, consolidating Unions, LetRec consolidations, and index exports. Introspection dataflows keep plain batchers. The near/far threshold reaches batchers through a process-global installed in apply_worker_config from compute_temporal_bucketing_summary (shared with the operator). Also register the flag where bin/lint-test-flags requires it: in parallel-workload's FlipFlagsAction and in the CI system parameters, on by default in get_minimal_system_parameters (all CI runs exercise batcher mode, version-gated) and variable in get_variable_system_parameters (parameter-randomized runs cover the flag-off configuration). Co-Authored-By: Claude Fable 5 <[email protected]>
Extend test/sqllogictest/temporal_bucketing.slt with an end-to-end section running temporal-filter workloads in batcher mode: Reduce, TopK, EXCEPT ALL (consolidating Union plus Threshold), and an indexed view, with timestamps far in the past and future relative to any plausible test-run time so results are deterministic. A second section runs with enable_column_paged_batcher also enabled, covering the paged temporal batcher in an indexed view and a self-join. Since CI otherwise runs with the batcher flag on, add temporal_bucketing_legacy.slt covering the flag-off configurations that exist in production while the flag rolls out: the standalone operator (batcher flag off, operator flag on) and plain batchers (both off). The file documents that it should be deleted when batcher-mode bucketing becomes unconditional, and the two flags' doc comments point at it so its deletion is not forgotten. Co-Authored-By: Claude Fable 5 <[email protected]>
ggevay
force-pushed
the
temporal-bucketing-merge-batcher
branch
from
July 12, 2026 16:04
1a3c8f8 to
9934d19
Compare
ggevay
marked this pull request as ready for review
July 12, 2026 16:48
Contributor
Author
|
We decided with @antiguru that we'll get back to this once |
bosconi
reviewed
Oct 28, 2026
bosconi
left a comment
bosconi
left a comment
Member
There was a problem hiding this comment.
Notes for when this is picked back up, since it is parked until ColumnMergeBatcher is used everywhere. It conflicts with main in 13 files.
- With the flag off, four consolidation sites change exchange hash and operator name, which contradicts the bit-for-bit claim (inline).
rust-miri-test-fullin nightly 17162 fails oncolumn_pager::tests::align_variant_fast_pathbecause of this PR (inline).
Comment on lines
+185
to
+196
| pub(crate) fn consolidate<'s, D, R>( | ||
| &self, | ||
| collection: VecCollection<'s, T, D, R>, | ||
| must_consolidate: bool, | ||
| name: &str, | ||
| ) -> VecCollection<'s, T, D, R> | ||
| where | ||
| T: MaybeTemporalArrange, | ||
| D: ExchangeData + std::hash::Hash + Columnation + Ord, | ||
| R: ExchangeData + Semigroup + Default + Columnation, | ||
| { | ||
| T::consolidate_maybe_temporal(collection, must_consolidate, name, self.temporal_batcher) |
Member
There was a problem hiding this comment.
This always routes through consolidate_named_if, but four call sites replaced consolidate_named, which exchanges on Hashable::hashed() and names its operator Unpack {name}, where consolidate_named_if(true, ..) uses the fixed-seed AHash and unpack consolidated (operator.rs). So with the flag off, LetRecConsolidation, UnionConsolidation and both TopK final consolidates shuffle keys to different workers and rename in introspection.
Add a consolidate_named-shaped sibling on MaybeTemporalArrange / Context for those four sites to keep the off path identical.
| @@ -759,13 +777,19 @@ mod tests { | |||
|
|
|||
| #[mz_ore::test] | |||
Member
There was a problem hiding this comment.
This fails rust-miri-test-full in nightly 17162 on madvise: the 1024-element column always covers a whole aligned page, so pageout_swap now calls it. The sibling Swap-paging tests already carry the ignore.
Suggested change
| #[mz_ore::test] | |
| #[mz_ore::test] | |
| #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `madvise` on OS `linux` |
| assert_eq!(b.held_records(), 0); | ||
| } | ||
|
|
||
| #[mz_ore::test] |
Member
There was a problem hiding this comment.
This took 425s under miri in nightly 17162, the slowest test in the job by far.
Suggested change
| #[mz_ore::test] | |
| #[mz_ore::test] | |
| #[cfg_attr(miri, ignore)] // too slow |
This branch has not been deployed
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.
https://linear.app/materializeinc/project/clean-up-temporal-bucketing-dc1c6f9fcd84/overview
Moves temporal bucketing from the standalone
temporal_bucketoperatorinto the merge batcher itself: each batcher decides at every seal, based
on the data it actually holds, whether to park far-future updates in a
bucket chain. This localizes the bucketing complexity in the batcher
instead of smearing it across the lowering, covers every batcher site
uniformly, and also catches future updates no static analysis can see
(e.g. worker skew and input skew in multi-input operators in lagging dataflows).
Takes over and productionizes #36427.
The design doc in this PR
(
doc/developer/design/20260711_temporal_bucketing_batcher.md) has themotivation, the design, and the performance story. The PR is structured
for commit-by-commit review, and the commit messages carry the
per-commit details.
Rollout: everything is behind the new
enable_compute_temporal_bucketing_batcherflag, default off. Off meansbit-for-bit today's behavior. When on, it supersedes
enable_compute_temporal_bucketing(the operator is suppressed). The oldcode, the operator and its lowering plumbing, is not deleted here. That
is a later cleanup PR, once the rollout completes.
Engines: the primary focus is the new
ColumnMergeBatcher(thepaged engine), as discussed. The batcher is generic over its engine, and
a small adapter (~8% of the diff, will be deleted together with the old engine)
also covers the columnation batcher, for three reasons: the columnation
engine is currently the default everywhere, so the rollout can proceed
independently of the paged rollout. Paged builders exist only for the
row-row and val-row arrangement families, so uniform coverage of every
batcher site is only expressible on the columnation engine today. And CI
currently runs the columnation engine nearly exclusively, so the temporal
code paths would otherwise go largely unexercised.
Changes from #36427:
MergeBatcher, the batcher is a standalone type generic over aself-contained
TemporalMergerengine trait, reusing therecord-walking loops we already own in-repo.
ColumnMergeBatcherchains(primary, spillable far tail) and the columnation
ColInternalMerger(rollout carrier).
compute_temporal_bucketing_summary): data within θ of the frontierstays in the flat chains exactly like the plain batcher's kept
updates. The draft bucketed all future data. This is what makes
always-on safe for data that races a tick ahead of the frontier
without any temporal filter.
unknown-with-scan-fallback state): silent performance-degradation
bugs become loud invariant violations. Bounds are conservative under
consolidation, and a bucket whose contents cancel entirely drops its
content (a real bug class the protocol proptest caught: a fully
cancelled bucket retaining stale bounds panicked downstream trace
imports).
frontier never falls below a sealed upper (capability/hint protocol),
with a cheap runtime guard and an arrange-protocol proptest (with
cancelling updates) pinning it. The reported frontier is the attained
minimum of held times rather than coarsened to bucket boundaries.
ensure_collections; this PR covers every rendering batcher site ontotally ordered timestamps (ArrangeBy ok+err, Reduce, TopK,
Threshold, linear-join stages, consolidating Unions, LetRec, index
exports) and suppresses the operator at all of its insertion sites.
operator flag, both mechanisms coexisting, CI system-parameter
registration (batcher mode on in minimal, variable in randomized),
and a dedicated flag-off sqllogictest so the production
configurations stay covered while CI runs batcher mode.
BatcherEventaccounting(the draft's logger was dead), plus a cumulative records-touched
counter, a per-seal high-water mark, and a debug log line for seals
that did bucket work.
arrange-protocol proptest, paged variants under an always-resident
and a page-everything pager, and a criterion suite asserting parity
with the plain batcher on no-future and raced-ahead workloads and
the order-of-magnitude win on far-future ones.
installed from the worker config (live-tunable for new dataflows),
read once per batcher.
PagedColumnrecords its record count in pageout metadata, so heldrecords are accountable without materializing chunks.
extract's single comparison (classification tests the emitted part
first), bounds tracking is skipped where the seal never uses it, and
the near/far output builders allocate lazily. Criterion-gated at a
0.98-0.99 temporal-to-plain runtime ratio.
doc ("The first seal" section) argues the halving cascade (about 2 F
record-touches for interval-shaped tails) with simulations confirming
2-4 F across data shapes, the seal budget defers restore work across
subsequent seals so only the peel share runs synchronously, and a
regression test asserts the touched-records counter stays within
bounds for a hydration-shaped seal.
the performance regimes with per-regime baselines, and the rollout
plan.
Validation: full Nightly runs are green in batcher mode (CI runs
with the flag on everywhere via the minimal system parameters). A sweep
of all 90 feature-benchmark scenarios against the merge base measured
overall parity (median runtime ratio 1.0025), with the temporal-filter
scenarios at parity against today's operator mode.
Nightly runs:
🤖 Generated with Claude Code