Repository navigation
SS-454 - storage: group a pinned snapshot's updates into one batch - #38935
Conversation
fc77c10 to
05deedd
Compare
1b5035a to
a4b2bf8
Compare
eeed893 to
7c35274
Compare
7c35274 to
c7dd0cf
Compare
f460ce5 to
99d7e35
Compare
99d7e35 to
b737542
Compare
QA LLM Review1. MEDIUM -- Finishing uncovered builders on description arrival splits a timestamp into two batches, with the flag off
DetailsA writer can pick up the description before a timestamp's trailing rows whenever its future is mid-await while both are queued. For example, it may be waiting on I reproduced this with the PR's Finishing early doesn't save meaningful memory either. The minter mints |
b737542 to
f220215
Compare
confirmed and addressed |
| /// lookahead costs: once the snapshot ends the shard upper waits for the frontier to reach the | ||
| /// ceiling, which is about one lookahead regardless of how long the snapshot ran. | ||
| /// | ||
| /// Zero disables committing ahead, leaving descriptions derived from the frontier alone and every |
There was a problem hiding this comment.
Checking my understanding:
This is not the same as zero lookahead, which would be "max timestamp seen in the data" plus zero.
And while we're hydrating, the frontier is stuck at 0 but "max timestamp seen" is whatever we've read from the CDC stream.
There was a problem hiding this comment.
correct, a zero here translates to "don't lookahead". Part of that is ConfigVal doesn't have support for an Option<Duration> (and I didn't want to add it). But more importantly, if zero was used, it would just clamped to at least the tick interval (fn description_lookahead in persist_sink.rs)
| /// milliseconds, the unit of `mz_repr::Timestamp`. `None` turns committing ahead off. | ||
| /// | ||
| /// Floored at `timestamp_interval`, since a ceiling that does not clear a timestamp tick is behind | ||
| /// the data by the time it reaches a writer. |
There was a problem hiding this comment.
I'm not sold on this statement.
Pedantic take follows, but it's the result of me trying to understand why this doc comment confused me so much.
-
I think it's possible for a worker to process a ceiling at X before the rows at X-1 depending on how those rows got distributed across writers. (Maybe all the earlier rows at time X-1 were sent to other writers.)
-
Similarly, is it possible for a ceiling at X to be delayed long enough for a worker to have already seen rows at X, X+1, X+2, etc?
| /// the data by the time it reaches a writer. | |
| /// We'd like a ceiling to include the data it's derived from. | |
| /// We set the minimum lookahead to one timestamp tick so that if the writer | |
| /// processes the ceiling and the data in the same round, | |
| /// the data will be included under the ceiling. | |
| /// Shorter lookaheads generate ceilings that likely (always?) exclude their corresponding data. |
EDIT: I'm not really sure what this comment should say. I'm not sure what it would mean to have e.g. 0.5tick lookahead
There was a problem hiding this comment.
The comment is just calling out that if the lookahead is less than the tick interval, then we would almost certainly get data that was beyond whatever ceiling was set.
1. I think it's possible for a worker to process a ceiling at X before the rows at X-1 depending on how those rows got distributed across writers. (Maybe all the earlier rows at time X-1 were sent to other writers.)
I might be misunderstanding the statement, but I would say that's the expected mode. A writer operator observes a max_seen_timestamp X as the new ceiling, but will see data at timestamps < X.
2. Similarly, is it possible for a ceiling at X to be delayed long enough for a worker to have already seen rows at X, X+1, X+2, etc?
If the lookahead is small enough, then it is possible that the max_seen_timestamp exchange took more rounds to resolve than it did for the data to make it to a writer. This isn't a correctness issue, it just means that persist sink would create extra builders for the data that arrived beyond the ceiling.
The minimum of a timestamp tick seemed the most logical because it allows testing at the boundary condition, but we could make the default some multiple of the tick rate instead. For production, we'd set lookahead on the order of 5s-10s, which is 5-10 times larger than the default tick interval.
There was a problem hiding this comment.
Updated the comment based on our conversation. Let me know if this reads better!
| /// Passes `desired` through, and on a second output reports the largest timestamp this worker's | ||
| /// share has reached. | ||
| /// | ||
| /// `desired` is pre-sharded, so any one worker sees a share of it. The batch description minter |
There was a problem hiding this comment.
I read this as "before sharding" 😅
But actually it's "sharded before".
3475421 to
98979c4
Compare
QA LLM Review1. MEDIUM -- Pacing the ceiling on the remap upper stretches the tail to the whole post-snapshot catch-up for MySQL and SQL Server
MySQL and SQL Server keep minting bindings during a snapshot, but their replication only starts after it. So the last ceiling ends up a lookahead past the wall-clock time the pin lifts, not past the export's data. The export's shard upper then stays at its as_of until replication has replayed every upstream change made during the snapshot. The design doc and PR description promise a tail of about one lookahead, and with the flag off or with the parent's DetailsThe MySQL statistics operator and the SQL Server progress operator probe on their own ticker, outside the busy signal ( The pin lifts only once replication passes the snapshot point. For MySQL that is the rewind caps ( Example: a one-hour MySQL snapshot on a server with steady writes, with a 5s lookahead, keeps the upper at the as_of until the whole hour of binlog has been replayed. With the parent it stayed there for about 5s past the rewind point. PostgreSQL is barely affected. It only probes once before the snapshot ( |
| /// The next thing the minter emits, or `None` when there is nothing to do. | ||
| /// | ||
| /// A description is derived from the frontier, which certifies that everything below it has | ||
| /// arrived. An outstanding `committed` ceiling suppresses that until the frontier reaches the | ||
| /// ceiling, which is what collapses a whole snapshot and the catch-up behind it into one | ||
| /// description. | ||
| /// | ||
| /// A `lookahead` is given only while a snapshot pins the frontier, and commits a ceiling that far | ||
| /// past `remap_upper`. Every reclocked update is stamped below the remap upper before it exists, | ||
| /// so the ceiling leads every row that can still arrive, and the lead is what lets it reach the | ||
| /// writers first, since a builder only takes updates at times it was opened for. | ||
| /// | ||
| /// Committing is confined to a snapshot because a ceiling is binding. While it is outstanding no | ||
| /// description is derived from the frontier either, so once the snapshot ends the shard upper waits | ||
| /// for the frontier to reach the ceiling rather than advancing to where the frontier actually is. A | ||
| /// collection that is keeping up lags the data by about one timestamp and so has nothing to group, | ||
| /// and would pay that wait for nothing. |
There was a problem hiding this comment.
I was struggling to understand the comment and the code for next_mint,
so I tried thinking about what it must be and came up with this:
/// `current_upper`:
/// Data shard upper. The upper of the previous batch.
/// We cannot mint an upper <= this timestamp (because it's already been minted).
/// `desired_frontier`:
/// Data input frontier.
/// If we mint a new upper, the new upper will be `desired_frontier`.
/// `remap_upper`:
/// Remap shard upper. Constantly advancing.
///
/// `committed`:
/// We cannot mint an upper < this timestamp (because we're building a consolidated batch).
/// `lookahead` (Optional):
/// If present, we assume `desired_frontier` is pinned (e.g. due to snapshot),
/// so we advance `committed` as `remap_upper` + `lookahead`
/// to consolidate in-flight data into one shared batch.
Does this explanation make sense to your brain as well?
There was a problem hiding this comment.
Aside from being thrown off by the idea of minting uppers, seems reasonable! 😂
I might adjust these two descriptions:
current_upper: the previous value of desired_frontier, initialized from the shard
remap_upper: the upper of the remap shard, which advances on each successful probe
If we mint a new upper, the new upper will be
desired_frontier.
I think this is a little confusing, but maybe you mean advanced the data frontier? It isn't to do with minting a binding, at least not directly. This frontier advances when the data frontier (the stream output by the source reader), translated to IntoTime, moves past the remap trace since (which is the earliest binding reclock knows about).
We cannot mint an upper < this timestamp (because we're building a consolidated batch).
This sounds accurate. I'd say we should mention the description (since that's what gets minted), i.e. "the minter will not mint a batch description with an upper less than this while snapshotting". This is also the ceiling, and determines which data are put into the consolidate batch.
There was a problem hiding this comment.
Incorporated into what you proposed:
/// `current_upper`:
/// The previous value of `desired_frontier`, initialized from the shard
/// Any data before this has already been processed.
/// `desired_frontier`:
/// Data input frontier.
/// This frontier advances when the reclocked data stream advances.
/// `remap_upper`:
/// The upper of the remap shard, which advances on each successful probe.
///
/// `committed` (Optional):
/// When set, the minter will not mint a batch description with an upper less
/// than this while snapshotting. This is also the ceiling, which determines
/// which data are written to the consolidated batch.
/// `lookahead` (Optional):
/// If present, `desired_frontier` is pinned to `as_of` (e.g. due to snapshot),
/// so we advance `committed` as `remap_upper` + `lookahead`
/// to consolidate in-flight data into one shared batch.
There was a problem hiding this comment.
Yeah, I meant minting the upper of the batch. I noticed Mint::Description only carries the upper bound of the batch, which makes sense since batches are in order and have no gaps.
I did miss the bit about minting the combined batch specifically at the ceiling (rather than at desired_frontier, per steady state), which makes sense because everyone downstream has been building a batch up to that ceiling specifically.
I like what you've got there 👍
There was a problem hiding this comment.
Wait, I'm confused again. The only place we return Mint::Description, we use desired_frontier, not the actual committed ceiling.
There was a problem hiding this comment.
I wasn't confused about that part. The part I meant was:
desired_frontier does not advance the same way as committed.
We choose the batch upper to mint based on desired_frontier, not committed.
So I think the original note for desired_frontier is truer to what we're doing:
/// `desired_frontier`:
/// Data input frontier.
/// If we mint a new upper, the new upper will be `desired_frontier`.
modulo terminology about upper/batch/description/mint
There was a problem hiding this comment.
Checking that I understand what's happening:
// We want to `Mint::Description(desired_frontier)`.
// We can only do this if both `current_upper` and `committed`
// allow us to mint a description there.
if PartialOrder::less_than(current_upper, desired_frontier)
&& committed
.is_none_or(|c| PartialOrder::less_equal(&Antichain::from_elem(c), desired_frontier))
{
return Some(Mint::Description(desired_frontier.clone()));
}
// If `desired_frontier` is pinned, we advance the committed ceiling with `remap_upper`.
// Do nothing if `desired_frontier` is not pinned.
let lookahead = lookahead?;
// Do nothing if the data shard is closed.
let lower = *current_upper.as_option()?;
// Do nothing if the remap shard is closed.
let ceiling = remap_upper.as_option()?.checked_add(lookahead)?;
// Advance the committed ceiling if the batches haven't already passed it.
(lower < ceiling && committed.is_none_or(|c| c < ceiling)).then_some(Mint::Ceiling(ceiling))There was a problem hiding this comment.
Pretty much!
"Do nothing" in this case means returns "None" (at the function level), at the operator level, None means the minter doesn't emit a ceiling update or a description.
// Advance the committed ceiling if the batches haven't already passed it.
(lower < ceiling && committed.is_none_or(|c| c < ceiling)).then_some(Mint::Ceiling(ceiling))
When you say the batches haven't already passed it (the ceiling), I'm not following. This operator doesn't know what batches exist. I would read this line as "return the new ceiling if it is beyond the last known data frontier and it is beyond the existing ceiling" - where an unset ceiling is less then everything.
There was a problem hiding this comment.
Nice. So that last bit is the one I wasn't sure about earlier (and the reason my attempted explanation makes no sense): Why do we have the lower < ceiling check?
There was a problem hiding this comment.
As I was updating per our review yesterday, I dug into this a little further and finally remembered. The operator calls this function in a loop. So when adding a new table to an existing source, the dataflow restarts, and the first thing reclock does is downgrade to the as_of. persist_sink will see that fontier, and the minting operator loops on while let Some(mint).... On the first iteration, the current_upper gets bumped, but the operator's remap_upper is still Timestamp::minimum. On the next, we have to account for that with lower < ceiling. On the third, next_mint returns None and the operator yields.
Confirmed. Updating the oltp_sources to only include PG for now. MySQL and SQL Server will be added back in when concurrent snapshot/CDC are implemented for them. |
| committed: Option<mz_repr::Timestamp>, | ||
| lookahead: Option<u64>, | ||
| ) -> Option<Mint> { | ||
| let frontier_reached = |ts: mz_repr::Timestamp| { |
There was a problem hiding this comment.
OH
I did it again. For the past day at least, I've been wondering about the naming of this predicate.
I read it as, "Frontier reached by the argument!"
But actually, it's "Frontier reached the argument!"
| current_upper.clone_from(&desired_frontier); | ||
| // The description that retires a ceiling covers everything the writers grouped | ||
| // under it, so the next one starts fresh. | ||
| committed = None; |
There was a problem hiding this comment.
This looks like the right thing to do, but how did you find it?
I would've thought it doesn't matter unless the input frontier could be pinned multiple times during the lifetime of the operator.
Otherwise, it's okay that committed stays somewhere in the past and is effectively ignored in next_mint. (Is that true?)
There was a problem hiding this comment.
Actually, scratch that. committed only stays in the past because lookahead never becomes Some(_) again.
So IIUC it doesn't matter that we set committed = None here. Even if we could pin again, when lookahead is turned on, it will pull committed forward regardless.
There was a problem hiding this comment.
I found an obvious answer.
I didn't realize the Commitment stream was (lower, ceiling), where lower acts like an ID for the record. And once we emit a Batch Description, we no longer have a Commitment(lower, ceiling) record in hand, so we might as well say None (even if it's already implied by the outdated ceiling value itself).
I still don't know why we expect to have multiple lowers in the Commitment collection, but maybe I'll figure that out as I get deeper into this PR.
| // A commitment for a different lower belongs to a later description, so it | ||
| // cannot widen the open builder, which is finished under its own. | ||
| match commitment { | ||
| Some(held) if held.lower == next.lower => { | ||
| commitment = Some(Commitment { | ||
| ceiling: held.ceiling.max(next.ceiling), | ||
| ..held | ||
| }) | ||
| } |
There was a problem hiding this comment.
My understanding (and 2 of 2 Claudes agree) is that since next_mint only emits ceilings while there's a lookahead due to a pinned data frontier during snapshot processing, it's not actually possible for commitments to have different lowers.
There was a problem hiding this comment.
yeah, good point. I'll clean up the comment, and panic on the lower changing. I didn't start with an explicit "set the lookahead if we are snapshotting", so this is unfortunately left over!
There was a problem hiding this comment.
ah - there's still a reason to keep this! When a new table is added, the minter emits 2 things: a description covering (0, as_of) and Commitment{ lower:as_of }.
| // The description that retires a commitment ends at or past its | ||
| // ceiling, so rows landing between its arrival and its readiness can | ||
| // still join the builder it will be finished under. |
There was a problem hiding this comment.
I wasn't sure what each "it" was:
| // The description that retires a commitment ends at or past its | |
| // ceiling, so rows landing between its arrival and its readiness can | |
| // still join the builder it will be finished under. | |
| // The description that retires a commitment ends at or past the commitment's ceiling, | |
| // so rows landing between the description arriving and the description becoming ready | |
| // can still join the builder that will finish under that description. |
I also couldn't understand what this comment is saying. I think it's trying to convince me that some assumption I have is incorrect. My guess is that I'm not making the assumption, and that's why I don't get this comment. Or there's some problem I'm not aware of that it's addressing (which is something I should figure out so I can review it properly).
There was a problem hiding this comment.
- We know there's at most one commitment.
- We specifically construct batch descriptions so the entire commitment ends up in one batch.
So isn't it just obviously okay to expand the commitment like this?
There was a problem hiding this comment.
obvious after enough time, the comment just helps with that.
f36c9a7 to
49aefe2
Compare
While an export snapshots, its frontier sits at the as_of, so the minter derives no description and every update writes a batch of its own timestamp. The minter now commits a ceiling a fixed lookahead past the data and broadcasts it to the writers, which group everything below it into one builder. The ceiling binds the minter to mint nothing below it, so the snapshot and the catch-up behind it become one description, appended once. The cost is a tail of about one lookahead before the shard upper catches up. Only exports that snapshot in this incarnation take part. The lookahead defaults off in production and on in the mzcompose configuration. Unit tests added. Co-Authored-By: Claude <[email protected]>
It's possible that an upstream table has no updates for a period exceeding the lookahead window, which would result in extra batches. This change derives the ceiling from the remap upper, which progresses regardless of data.
49aefe2 to
a65a199
Compare
| // A lower at the ceiling, as when the current upper jumps to an as_of ahead of the observed | ||
| // remap upper, leaves an empty range to commit. | ||
| assert_eq!( | ||
| next_mint( | ||
| &frontier(15), | ||
| &frontier(15), | ||
| &frontier(5), | ||
| None, | ||
| Some(LOOKAHEAD) | ||
| ), | ||
| None | ||
| ); |
|
thanks for the review! |
The storage persist sink has been updated to support grouping data into a single batch by committing to an upper while the the frontier is pinned to the snapshot time. This affected the benchmarks, which detected a ~5s increase in wallclock for tests that covered snapshot. This PR lowers the CI lookahead to 2s (to make it faster, while still offering coverage). It also moves the version gate from v26.41.0-dev to v26.45.0-dev, the first release containing the dyncfg (#38935). Before this, mzcompose set the parameter on v26.41 to v26.44 images that don't know it, which affects upgrade and platform-check runs. The v26.45 benchmark baseline now runs with the same lookahead as HEAD. Closes [SS-500](https://linear.app/materializeinc/issue/SS-500)
While an export snapshots its frontier is pinned, so no description gets minted
and every update writes a batch of its own timestamp.
The minter now commits a ceiling a fixed lookahead past the newest data and
broadcasts it on a second output.
write_batchesputs everything below it intoone builder, whatever the timestamp.
A ceiling isn't a description. It carries the lower of the description that'll
retire it, since a builder declares its lower when it opens, and it binds the
minter to mint nothing below it. So the snapshot plus its catch-up is one
description, one append. The cost is the tail, the shard upper waits for the
frontier to reach the ceiling, about one lookahead however long the snapshot ran.
Rows that outrun the ceiling write their own batch, same as everything does with
the flag off. Both kinds get finished under whatever description covers them.
max_seen_timestampsreports each worker's largest timestamp to the minter,since the collection is pre-sharded and the minter would otherwise pace on its
own share.
Only the OLTP sources hold a snapshotting export's frontier at the as_of, so
only their exports take part, and only while the frontier sits there.
storage_persist_sink_description_lookaheaddefaults to 0, off in production,5s in mzcompose.
Unit tests added.
🤖 Generated with Claude Code