Repository navigation
Conversation
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]>
…ing an empty chain 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]>
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
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.
Spike of a leaner alternative to #37589: temporal bucketing as a thin wrapper around the existing merge batchers instead of a rewritten batcher.
TemporalBatcherroutes updates at or beyondmax(frontier edge, as_of) + summaryinto 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 smallTimedUpdatestrait, and over iterative timestamps by bucketing on the outer coordinate against frontier elements whose inner time is the minimum.The wrapper replaces the
temporal_bucketoperator 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
Producttimes, spilling, stepped hydration uppers) and a criterion bench inmz-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