Repository navigation
compute: spill correction chunk bodies through the buffer pool - #38897
Conversation
antiguru
changed the title
47662fb to
d789f1f
Compare
d789f1f to
5a9f690
Compare
antiguru
left a comment
antiguru
left a comment
There was a problem hiding this comment.
Self-review.
- Gate: correction spilling turns on when either
enable_column_paged_batcher_spillorenable_upsert_paged_spillis on. Enabling the upsert flag therefore starts spilling every MV sink's correction buffer in the process. This coupling across teams should not live only in a flag description. The correction buffer needs its own flag, default off in production and on in CI. correction_mem: the--pagerpresets still callcolumn_pager::apply_tiered_config, which neither buffer reads after this PR. Also,--chunk-sizedefaults to 8192, while the flag now defaults to 2 MiB.- Dead API:
Spilled::compressedhas no reader outsidechunk.rs, sotry_spill_refcan return the handle directly. - Nits: neither
#[inline(always)]onColumn::borrownor thepubonat_serialized_capacityhas a user or a measurement behind it.
🤖 Posted by Claude Code
5a9f690 to
b97665a
Compare
|
Addressed:
🤖 Posted by Claude Code |
| /// cleared once populated, so its contents have a stable address. | ||
| /// cleared once populated, so its contents have a stable address. Taking the body frees the | ||
| /// pool's copy, so the chunk holds exactly one. | ||
| fn column(&self) -> &Column<(D, Timestamp, Diff)> { |
There was a problem hiding this comment.
column frees the pool slot on any read, so a chunk that's read and kept stays on the heap for good. That's fine for merge inputs, but not for emitted. The persist write reads all of it, and it then sits in the buffer until the feedback comes back and the next read merges it away. So a whole batch, the full snapshot right after hydration, can't be spilled at the point where we most need it to be.
The pool already has a read that doesn't free the slot (read_into), and ColumnChunk uses it for with_column, keeping into_column as the consuming read. Can Chunk do the same? Merges would take the body, and everything else would read a copy. Not a regression, since the pager also took the body on first read, but this PR seems like the place to fix it.
There was a problem hiding this comment.
Agreed, done in this PR. Chunk::with_view reads a materialized chunk in place. Otherwise it copies the body out of the pool with read_into for the length of the call, so the slot stays and the pool can still spill it. Chain::iter uses it, and so does the soft-assertion tie check in can_accept_chunk, which used to materialize both chunks. iter turns one chunk's updates into owned values at a time, so the iterator holds at most one chunk's worth. The soft assertion on emitted in updates_before now reads the cached last time and touches no body.
Only merges and splits take the body now. They go through Chunk::column, and the pool round-trip test checks that every body is still pooled and none is materialized after iter. In correction_mem at a 64 MiB pool budget, drain takes 4.7 s and live RSS ends near 30 MiB.
🤖 Posted by Claude Code
b97665a to
ba73103
Compare
| } | ||
|
|
||
| /// Reads a `/proc/self/status` size field, in MiB. | ||
| fn status_mib(field: &str) -> f64 { |
There was a problem hiding this comment.
Yes. It reads /proc/self/status. Off Linux it now prints NaN for the memory figures instead of panicking, so the timings still work, and the module doc says so.
🤖 Posted by Claude Code
|
We should benchmark with this enabled at some point |
DAlperin
left a comment
DAlperin
left a comment
There was a problem hiding this comment.
LGTM mod a nit and q, thanks!
`CorrectionV2` paged its chunk bodies through the column pager, a separate spill substrate with its own policy, backends, and accounting. Chunk bodies now go to `mz_ore::pool`, which is where the rest of the codebase spills and which accounts for its own residency. Under a 128 MiB cap with swap the pool is 1.13x faster than the best tuned pager configuration and needs no compression to get there, and at a 64 MiB budget it holds live RSS at a third of the unspilled figure. The spill decision moves into one function in `mz_timely_util::columnar::chunk`, shared with `ColumnChunk`, so the gates, the installed pool, and the size floor live in one place. A new `try_spill_ref` spills from a borrowed column and encodes straight into the pool slot, so a spilled body costs no intermediate allocation and the caller keeps its column allocation to refill. It reads a gate of its own, set by the new `enable_compute_correction_v2_spill` flag, so turning on the arrange or upsert spill flags does not change how MV sinks use memory. The flag defaults off in production and on in CI, and turning it on installs the pool like the other two spill flags. `ChunkBuilder` no longer wraps `ColumnBuilder`: it owns one column, mints as soon as the column reports itself at capacity, and returns the minted chunk from `push`, which retires the builder's pending queue. Merges and splits take a chunk's body out of the pool, freeing the slot, so a chunk holds exactly one copy of its body at any time. Other reads copy the body out for the duration of the read and leave it in the pool. The main one is the persist write reading `emitted`, so a written batch, as large as the snapshot right after hydration, stays spillable until the next consolidation merges it. A body the pool declines is serialized into an exactly sized allocation rather than kept as a typed container, whose growth slack costs 54% of peak RSS. The `correction` bench gains a spill sweep. The `column_borrow` bench's encode group compares minting into a fresh allocation against a reused one, which is the cost `try_spill_ref` avoids. The `correction_mem` example runs one fill per process and reads `VmHWM` and `VmRSS`, which criterion cannot do because setup, measured closure, and the previous iteration's drop overlap. Co-Authored-By: Claude Fable 5.1 <[email protected]> Co-Authored-By: Claude Opus 5.5 <[email protected]>
ba73103 to
ba795f4
Compare
|
Benchmarked on staging, with the stack deployed. The workload creates a materialized view
With spilling on, each run inserted about 40,000 bodies into the pool, wrote 32 to 36 GB of extents, and peaked at about 6 GB of swap. Spilling costs no wall-clock time and about 8% more CPU. Peak memory went down, though a 5 s sample is too coarse to put an exact number on it. A sweep over 100cc and 50cc, and the M.1 sizes with the same CPU and memory but more disk, is running now, and I'll post it here. 🤖 Posted by Claude Code |
`compute_correction_v2_chunk_size` sizes `CorrectionV2`'s staging area, not its chunks: chunk bodies are fixed at the columnar ship size. The staging size sets how many updates accumulate before a chain is minted, and every chain minted is a chain some later read has to merge. Raising the default from 8 KiB to 2 MiB, one chunk body's worth, takes insert on a 4.19M-update append workload from 3503 ms to 1897 ms, with drain and peak RSS unchanged. These are medians of three `correction_mem` runs with the rest of the stack applied and no pool installed. The curve is monotone across the range (the new `bench_chunk_size` sweep in the `correction` bench shows it), so the default is bounded only by the per-sink, per-worker cost of the staging vector. The stage counted its fill in tuples of `size_of::<(D, Timestamp, Diff)>()`, which is 40 bytes for a `Row` however wide the row is, so a byte-sized flag bounded neither the staging memory nor the reported size once rows spill out of line. A new `DataBytes` trait on the buffer's data types reports each datum's bytes with the heap included (`Row` via `byte_len`, `DataflowErrorSer` via its serialized length), the stage ships when the staged bytes reach the flag, and its size metrics carry the same figure. `stage_ships_by_bytes` checks that rows of two widths ship after the number of updates their bytes predict. That vector was allocated eagerly at the configured capacity and re-allocated at that capacity on every flush. At 2 MiB per sink and worker that is no longer free, so it now grows on demand and gives its allocation back on flush. The mzcompose and parallel-workload flag tables follow the new default, since CI runs the value listed there rather than the compiled default. The flag's description now says what it sizes. First of four, split out of #38645: this, the O(1) metrics (#38895), the per-chunk borrow hoist (#38896), and the pool substrate (#38897) land in that order. Supersedes the fork PR #38647. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Fable 5.1 <[email protected]>
|
Staging results across replica sizes, as promised above. The workload is the same as before: a materialized view V1 against V2, three sizes running in parallel in one environment, so both versions saw the same shared persist and S3 load:
Spill off against spill on, runs one at a time. The pool budget is the staging budget fraction of 5%.
A CPU profile of the 200cc runs attributes about 15% of replica CPU to the correction buffer. Persist encoding and decoding take most of the rest. 🤖 Posted by Claude Code |
`Column` served two roles. On a dataflow edge it is built by pushing, crosses an exchange as channel bytes, and is read on arrival, which is what its `Typed`, `Bytes`, and `Align` variants are for. Behind an edge, in a merge batcher's chains, a spill-backed `ColumnChunk`, and a batch builder's seal, the same type held sorted, consolidated bodies that were only ever typed or, when read back from a store, serialized words. That code never constructed `Bytes`, and it asserted the variant it expected at seven sites with `unreachable!`. `ColumnBody<C>` is the type at rest: `Typed(C::Container)` while a body is written, `Words(Vec<u64>)` once a store handed it back, and nothing else. Writing to a serialized body materializes it by a bulk per-leaf copy, so the merge, extract, advance, and settle bodies address the typed containers without matching on a variant. `ColumnChunk`, `ColumnChunker`, `ColumnMerger`, `UnchunkBuilder`, `RowRowColPagedBuilder`, and the correction buffer's chunks and builder now hold bodies. The edge container keeps its name and every operator that names it is untouched. The boundaries are moves. A chunker takes whatever the edge delivered and emits bodies, and a body goes back onto an edge through `From`, typed staying typed and words becoming `Align`. The column pager's batcher and `ValRowColPagedBuilder` keep `Column` chains until the pager is retired, so the merge and extract bodies are generic over a `MergeChunk` trait both types implement, and `PagedChunker` wraps the chunker for that batcher. Fifth in the stack, on top of #38897. The chunked consolidation from #38900 hands its bodies back onto the edge through `From`. #38334 touches the same files and will need rebasing onto this, since edge paging lands on `Column` alone, which is the point of the split. Part of [CPU-279](https://linear.app/materializeinc/issue/CPU-279). 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Fable 5.1 <[email protected]>
CorrectionV2paged its chunk bodies through the column pager, a separate spill substrate with its own policy, backends, and accounting. Chunk bodies now go tomz_ore::pool, which is where the rest of the codebase spills and which accounts for its own residency. Under a 128 MiB cap with swap the pool is 1.13x faster than the best tuned pager configuration and needs no compression to get there, and at a 64 MiB budget it holds live RSS at a third of the unspilled figure.The spill decision moves into one function in
mz_timely_util::columnar::chunk, shared withColumnChunk, so the gates, the installed pool, and the size floor live in one place. A newtry_spill_refspills from a borrowed column and encodes straight into the pool slot, so a spilled body costs no intermediate allocation and the caller keeps its column allocation to refill.The correction buffer spills under a flag of its own,
enable_compute_correction_v2_spill. It defaults off in production and on in CI, and turning it on installs the pool like the arrange and upsert spill flags do. Those two flags gate chunk spilling only, so enabling them does not change how MV sinks use memory.ChunkBuilderno longer wrapsColumnBuilder: it owns one column, mints as soon as the column reports itself at capacity, and returns the minted chunk frompush. Merges and splits take a chunk's body out of the pool, freeing the slot, so a chunk holds exactly one copy of its body at any time. Other reads copy the body out for the duration of the read and leave it in the pool. The main one is the persist write readingemitted, so a written batch, as large as the snapshot right after hydration, stays spillable until the next consolidation merges it. A body the pool declines is serialized into an exactly sized allocation rather than kept as a typed container, whose growth slack costs 54% of peak RSS.The
correctionbench gains a spill sweep. Thecolumn_borrowbench's encode group compares minting into a fresh allocation against a reused one, which is the costtry_spill_refavoids. Thecorrection_memexample stays an example: it runs one fill per process and readsVmHWMandVmRSS, which criterion cannot do because setup, measured closure, and the previous iteration's drop overlap.Supersedes the fork PR #38645, which holds the earlier review discussion.
🤖 Generated with Claude Code