storage/upsert: v2 stash on ChunkBatcher with pool-backed spill - #38255
Conversation
DAlperin
changed the title
9ff7737 to
bbe7496
Compare
bbe7496 to
5c9756b
Compare
…nkSpine with the UnloadChunk drain
5c9756b to
465a3f5
Compare
The seal NOTE described a pre-pass-through cost model: since the whole-chunk extract pass-through, only chunks the upper splits round-trip the pool codec. The DrainStats docs and the eligibility comment still referred to the removed cursor path. Add a drain test with more distinct eligible keys than one probe window holds, so the mid-chunk window boundary is exercised, and demote the per-worker spill-gate application log to debug.
9a5f824 to
94dccba
Compare
antiguru
left a comment
antiguru
left a comment
There was a problem hiding this comment.
Seems fine, left a comment inline. I didn't read src/storage/src/upsert_continual_feedback_v2.rs because I'm not familiar enough with it. I can if that'd help. Meanwhile, my bot's coming up with a review, too, so let's see what it has to say.
I'll leave approval to someone from the Storage team.
| let compute_spill = ENABLE_COLUMN_PAGED_BATCHER_SPILL.get(config); | ||
| let storage_spill = mz_storage_types::dyncfgs::ENABLE_UPSERT_PAGED_SPILL.get(config); | ||
| // Set compute's leg of the process-wide chunk spill gate. The | ||
| // gate ORs this leg with storage's, so chunks spill while either | ||
| // subsystem's flag is set. Storage's config application writes | ||
| // only its own leg, keeping the two flags from clobbering each | ||
| // other. | ||
| mz_timely_util::columnar::chunk::set_compute_spill_enabled(compute_spill); |
There was a problem hiding this comment.
Should've spotted it earlier, but I don't think there's much value of controlling compute and storage spilling independently. We don't (recommend) to co-locate compute and storage workloads, so we already effectively get this behavior for free. You could assume this in this PR, which I think would make some things easier?
antiguru
left a comment
antiguru
left a comment
There was a problem hiding this comment.
Mechanism looks sound and the arm split is clean. Findings inline; verification notes and the residual coverage gap below.
What I checked
Built the branch and ran cargo test -p mz-storage --lib upsert_continual_feedback_v2 (10/10 pass). On top of the flavor-equality harness I ran two throwaway differential tests:
- 3000 keys over 6 rounds, randomized inserts, deletes and same-time double-writes, with a lagging feedback loop. Flavors agree.
- 40000 wide-row keys, large enough to seal into several chunks, with a round of ineligible re-stashing and a third of the keys deleted. Flavors agree and both match a hand-computed expectation.
I also traced the probe-window boundary by hand for interleaved drop, eligible and ineligible records. It is correct: end lands on the first eligible record of the (PROBE_WINDOW + 1)-th distinct key, and a key's records are contiguous under the (key, ts) sort, so no key's eligible record is separated from its probe.
Coverage gap
set_spill_override is public and chunk.rs uses it, but the upsert harness never installs a pool, so no unit test drains a spilled body back through extract_into and the pool codec. That is the path this PR exists to add. The PR description's claim that "the paged arm is exercised by the same tests as the chunked one" holds for the representation but not for spilling. A third harness dimension, or one dedicated test, would close it.
I could not run that locally to confirm either way: mz_ore::pool::Pool::new() fails with ENOMEM from the mz-storage test binary in my sandbox, while mz-timely-util's own spilled tests pass. That looks environmental, so treat it as untested rather than as a suspected failure.
Cleared
batches_through(Antichain::new())is exactly whatTraceReader::cursordoes, so the chunked drain reads the same window as the paged one, including the pre-existing property that the snapshot is taken before the first await and misses batches landing mid-drain.- Compute's leg is set before the pool-install early return, and setting the gate with no pool installed is inert, since
spill_poolfalls through toactive_pool. - Storage's leg lands in
UpdateConfiguration, not a maintenance tick, so flips apply per config update. The two legs read different flags and write different atomics, so they cannot clobber each other. - Moving
enable_upsert_paged_spillout ofUNINTERESTING_SYSTEM_PARAMETERSmatches that file's own guidance. - Clippy is clean on
mz-storage,mz-computeandmz-storage-types. The twoextra_unused_lifetimeswarnings inmz-computepredate this branch.
Nits
The + Ord added to upsert_inner and build_upsert_operator is redundant: timely::progress::Timestamp already requires Clone + Eq + PartialOrder + Ord + Debug + Any + ExchangeData. Same for the ExchangeData + Clone + Debug + Ord repeats on the arm impls and the two drain functions.
One note on CI coverage rather than on the code: enable_upsert_chunked_stash defaults on in CI, but it is only meaningful under enable_upsert_v2, which is itself randomized and defaults off. Chunked-arm coverage therefore depends on the seed selecting v2. That matches the description, flagging it only so the expectation is explicit.
Posted by Claude Code
| for index in start..total { | ||
| let (key, ts, _diff) = view.get(index); | ||
| let ts = <T as columnar::Columnar>::into_owned(ts); | ||
| if persist_upper.less_equal(&ts) && !persist_upper.less_than(&ts) { |
There was a problem hiding this comment.
The eligibility test is written twice in two different spellings: here as less_equal(&ts) && !less_than(&ts), and in pass 3 (line 1210) as two early-continues. Nothing ties them together.
If they ever drift, pass 1 stops probing a key that pass 3 still classifies as eligible. The result is a missing retraction: no panic, no assertion, just wrong output. old_values.remove(key) returning None is indistinguishable from a genuine insert.
Suggest one helper returning a three-way classification (Drop / Eligible / Ineligible) and calling it from both passes, so the probe set and the classification cannot disagree by construction.
Posted by Claude Code
| batcher | ||
| } | ||
|
|
||
| fn flush( |
There was a problem hiding this comment.
This body is identical to ChunkedArm::flush (line 941) except for the single push line: std::mem::take(chunk) here versus ColumnChunk::from_column(std::mem::take(chunk)) there.
Worth collapsing into one shared free function parameterised by the push, or a trait method push_chunk(batcher, Column<..>) with flush provided as a default.
Posted by Claude Code
| // copy-out, per probed chunk, inside `extract_into`. | ||
| let batches = trace | ||
| .batches_through(Antichain::new().borrow()) | ||
| .expect("batches_through always succeeds for the empty upper"); |
There was a problem hiding this comment.
"always succeeds" overstates it. Spine::batches_through asserts on a closed trace rather than returning None, so the call can still abort. TraceReader::cursor's own panic text, "unable to acquire complete cursor for trace; is it closed?", is the accurate framing.
Posted by Claude Code
| // only its own leg, keeping the two flags from clobbering each | ||
| // other. | ||
| mz_timely_util::columnar::chunk::set_compute_spill_enabled(compute_spill); | ||
| if !(compute_spill || storage_spill) { |
There was a problem hiding this comment.
The change is right, but the commit message for it is not:
only storage wired its leg: enable_column_paged_batcher_spill installed the buffer pool without ever setting the gate
That sentence names storage and then describes compute's flag, and it is wrong about the starting state. On upstream/main, git grep set_storage_spill_enabled -- src/ outside chunk.rs returns nothing: neither leg was wired, so the gate was dead rather than half-wired. Worth rewording before merge, since the message is the durable record of why this leg exists.
Posted by Claude Code
The chunk spill gate is documented as the OR of a compute leg and a storage leg, but no production code drove the compute leg: enable_column_paged_batcher_spill only influenced buffer-pool installation, so it could not enable chunk spilling as the flag and module docs claim. Set the leg where pool config is applied. No behavior change with default flags: the gate is only consulted by the upsert-v2 chunk batchers, and both flags default off.
enable_upsert_chunked_stash (off in production, on and randomized in CI) selects between two instantiations of the upsert-v2 operator loop: * Chunked (flag on): ChunkBatcher stash, chunk-spine feedback arrangement, bulk-probe drain. Spills committed chunk bodies through the process buffer pool. * Paged (flag off, the default): paged columnar merge batcher stash, ValRowSpine feedback arrangement, cursor drain. Spills cold chains through the storage-owned column pager. The operator loop itself is shared: build_upsert_operator is generic over an UpsertStashArm, whose two impls carry the flavor-specific pieces (stash batcher, feedback spine, flush, drain). UpsertStashFlavor names the arms and resolves from the config set once, at operator construction, mirroring compute's ArrangementBatcher. Both flavors' spill paths stay gated by enable_upsert_paged_spill, so storage_state applies that flag to both the chunk gate and the restored upsert_stash_pager. The unit-test harness runs every scenario under both flavors and asserts they produce identical output.
Review feedback: * The drain's eligibility test was spelled twice, once in the chunked drain's probe-collection pass and once in its classification pass. Drift between them would produce silently missing retractions, since an unprobed key's absent prior value is indistinguishable from a genuine insert. Both passes, and the paged drain's cursor walk, now classify through one classify_time helper. * The two arms' flush bodies differed only in the push line. flush is now a provided trait method over a per-arm push_chunk. * Reword the batches_through expect: a closed trace fails the spine's internal assert rather than returning None, so 'always succeeds' overstated what the expect guards. * Trim bounds already implied by Timestamp (Ord, Clone, Debug, ExchangeData). * Add a test that forces committed chunks through a private buffer pool via the thread-scoped spill override, covering the seal and drain read-back of spilled bodies through the pool codec. The pool's insert counter asserts spilling actually happened.
94dccba to
6709a37
Compare
patrickwwbutler
left a comment
patrickwwbutler
left a comment
There was a problem hiding this comment.
LGTM as far as upsert goes - My main worry is correctness testing. This feels like the type of feature that we should really stress test with large sources to try and reveal behavior quirks. I'd say at the very least we should create a replica running this on the large sources in the qa_canary in the sandbox before I feel good flipping the switch anywhere in production. But, since it's all flag-gated, seems like prod risk is pretty minimal
|
@patrickwwbutler yep, agree that this is the plan! Will do plenty of replica testing before cutting over. TFTR! |
3 participants
Motivation
The upsert v2 (continual feedback) stash moves from its previous representation onto
ChunkBatcher/ChunkSpine: stashed updates live in columnar chunks that participate in the buffer pool, and the feedback arrangement drains throughUnloadChunkone chunk at a time instead of materializing whole batches. Withenable_upsert_paged_spillon, stash and arrangement state past the residency budget spills as compressed, budget-accounted extents rather than raw kernel-swapped heap.Benchmarked extensively on AWS (self-managed EKS, MSK source, swap-enabled nodes), August 2026 campaign:
The new representation is flagged:
enable_upsert_chunked_stash(off in production, on and randomized in CI) selects between theChunkBatcherrepresentation above and the previous paged-columnar-merge-batcher +ValRowSpinerepresentation, which stays the production default while the chunked flavor earns trust. The operator loop is shared; a smallUpsertStashArmtrait carries the flavor-specific pieces (stash batcher, feedback spine, flush, drain), and the arms are resolved from the config set once at operator construction, so a dataflow keeps its flavor for life. Spilling in either flavor remains gated byenable_upsert_paged_spill.Tips for reviewer
upsert_continual_feedback_v2.rs.enable_upsert_paged_spillis replica-scoped and composes with compute's spill gate as an OR on the process-wide pool.Checklist
$T ⇔ Proto$Tmapping (possibly in a backwards-incompatible way), then it is tagged with aT-protolabel.