Skip to content

compute: spill correction chunk bodies through the buffer pool - #38645

Closed
antiguru wants to merge 4 commits into
MaterializeInc:mainfrom
antiguru:moritz/correction-pool-substrate
Closed

antiguru wants to merge 4 commits into
MaterializeInc:mainfrom
antiguru:moritz/correction-pool-substrate

Conversation

@antiguru

@antiguru antiguru commented Sep 4, 2026 •

Copy link
Copy Markdown
Member

Stacked on #38649. Review the last commit; the diff shows #38647, #38648, and #38649 as well until they merge.

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.

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. Chunk::column takes its body out of the pool, freeing the slot, so a chunk holds exactly one copy of its body at any time. 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, and mz-timely-util a column_borrow bench for the per-element borrow and encode costs the design rests on. The correction_mem example stays an example: it 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.

🤖 Generated with Claude Code

@antiguru antiguru left a comment

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The individual changes look fine, but this should at least be three changes:

  • Changing the chunk size default.
  • mz-ore::pager integration
  • Metric accounting change

The examples are hard to re-use as-is. Could they be benches, too?

Comment on lines +1929 to +1937
#[inline]
fn refs_cmp<D: Data>(a: (Timestamp, Ref<'_, D>), b: (Timestamp, Ref<'_, D>)) -> Ordering {
#[inline]
fn cmp<'x, D: Data>(a: (Timestamp, Ref<'x, D>), b: (Timestamp, Ref<'x, D>)) -> Ordering {
a.cmp(&b)
}
cmp::<D>((a.0, D::reborrow(a.1)), (b.0, D::reborrow(b.1)))
}

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Between comment and function.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Moved refs_cmp below refs_eq with its own one-line doc, so refs_eq keeps the comment that explains the reborrow. Now part of #38649.

🤖 Posted by Claude Code

Comment thread src/compute/src/sink/correction_v2.rs Outdated
Comment on lines +2195 to +2196
// Columnation regions are not Stacked Borrows compliant: later pushes invalidate the
// provenance of previously stored items under Miri.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fine to ignore miri here, but we'ren't using columnation here anymore.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Kept the ignore, replaced the comment (the test exercises no unsafe code of its own and is slow under Miri). The three pre-existing tests in the module carry the same stale columnation comment; I left those untouched. Now part of #38648.

🤖 Posted by Claude Code

// pool stores it uncompressed and in its hottest eviction band.
// TODO: chunks resting in the far-future buckets outlive that assumption and want a
// depth that reflects how long they have gone untouched.
let (pooled, resident) = match chunk::try_spill_ref(column, 0) {

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we ever pass a non-0 depth? We could for deeper chains in the correction structure.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, always 0 today. Deriving it from the chain's merge generation (fresh chains 0, a merge output one deeper than its deepest input, splits inherit) is a small change, but the pool compresses at depth >= 1 (DEFAULT_COMPRESS_MIN_DEPTH), so it would put every merged chain's chunks behind lz4, and I have not measured that trade. Left as the TODO here; can follow up separately with numbers.

🤖 Posted by Claude Code

Comment thread src/compute/src/sink/correction_v2.rs Outdated
Comment on lines +1586 to +1588
// Serialize anyway. A typed container carries the slack of its last growth
// step, measured at 1.5x the serialized size of the bodies this builder mints,
// and a resident chunk holds whatever it is given for the rest of its life.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"Serialize into a vec to leave allocation in place, and avoid shipping spare capacity." Something like this?

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reworded along those lines: serialize into an exactly sized vector, which leaves the column's allocation in place and keeps the typed container's spare capacity out of the body.

🤖 Posted by Claude Code

Comment thread src/compute/src/sink/correction_v2.rs Outdated
Comment on lines +1635 to +1639
/// Any caller that touches more than one update must hoist this out of its loop and index the
/// returned view. `Column::borrow` rebuilds the struct-of-arrays view from the serialized
/// header on every call, measured at 7x the cost of indexing an already-borrowed view
/// (`mz_timely_util`'s `column_borrow_cost` example), and the sequential iterator it threads
/// through `FromBytes::from_bytes` blocks LLVM from hoisting the decode itself.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's avoid the measurement results, and just say it avoids the per-element borrowing cost.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Dropped the numbers and the example reference; the doc now only says that borrowing per element pays the header decode per element. Now part of #38649.

🤖 Posted by Claude Code

Comment thread src/compute/src/sink/correction_v2.rs Outdated
Comment on lines +1712 to +1715
/// Updates accumulate in one column, which is minted into a chunk as soon as its serialized
/// size reaches the ship granularity (~2 MiB, the same boundary the columnar merge machinery
/// ships at), so every chunk is a single, predictably sized body. Minting a spilled body leaves
/// the column's allocation in place, so a builder that mints many chunks grows one column once.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The 2MiB number is a property of the column builder, so just say we defer column formation to it.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reworded: the builder mints as soon as the column reports itself at capacity (Column::at_capacity), without naming the size.

🤖 Posted by Claude Code

Comment thread src/compute/src/sink/correction_v2.rs Outdated
Comment on lines +175 to +195
+ Columnar<Container: Send + Sync + Clone + for<'a> columnar::Borrow<Ref<'a>: Eq + Ord>>
+ Send
+ Columnar<
Container: Send
+ Sync
+ Clone
+ for<'a> columnar::Borrow<Ref<'a>: Eq + Ord, Borrowed<'a>: Send>,
> + Send
+ Sync
{
}
impl<D> Data for D where
D: differential_dataflow::Data
+ Columnar<Container: Send + Sync + Clone + for<'a> columnar::Borrow<Ref<'a>: Eq + Ord>>
+ Send
+ Columnar<
Container: Send
+ Sync
+ Clone
+ for<'a> columnar::Borrow<Ref<'a>: Eq + Ord, Borrowed<'a>: Send>,
> + Send

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Formatting looks off.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That was rustfmt giving up on the associated-type bound inside Columnar<Container: ...>. The container bounds now live in a DataContainer helper trait, which formats normally and gives the new Borrowed<'a>: Send bound a place to be explained. Now part of #38649.

🤖 Posted by Claude Code

@antiguru
antiguru force-pushed the moritz/correction-pool-substrate branch from d11c7c9 to 773a1d1 Compare September 4, 2026 12:19
@antiguru antiguru changed the title compute: fix the MV sink correction buffer's per-element decode, metrics walk, and spill substrate compute: spill correction chunk bodies through the buffer pool Sep 4, 2026
@antiguru

antiguru commented Sep 4, 2026

Copy link
Copy Markdown
Member Author

Split into four, landing in order: #38647 (staging size default), #38648 (O(1) metrics), #38649 (per-chunk borrow hoist), and this PR (pool substrate). Each is one commit on top of the previous; the diffs here are cumulative until the earlier ones merge.

The metrics change now keeps reporting capacity and allocations to introspection, which the introspection-sources.td failure in the last CI run was about.

The column_borrow_cost and column_alloc_cost examples became the column_borrow criterion bench in mz-timely-util (per-element borrow versus a hoisted borrow, and encode into a fresh versus a reused allocation). column_retain_cost measured RSS, not time, so it is dropped rather than made a bench. correction_mem stays an example because it reads VmHWM once per process, which criterion cannot express.

🤖 Posted by Claude Code

@antiguru
antiguru marked this pull request as ready for review September 14, 2026 18:06
@antiguru
antiguru requested review from a team as code owners September 14, 2026 18:06
@def-

def- commented Sep 14, 2026

Copy link
Copy Markdown
Contributor

QA LLM Review

1. MEDIUM -- compute_correction_v2_chunk_size sizes the stage in tuples, not bytes, so the 2 MiB default is a 256x staging-memory increase that scales with row width

src/compute/src/sink/correction_v2.rs:276

chunk_capacity = chunk_size / size_of::<(D, Timestamp, Diff)>() is a fixed 40 bytes for D = Row no matter how wide the row is, because Row is a 24-byte CompactBytes that heap-allocates past 23 bytes of payload. Raising the default from 8 KiB to 2 MiB therefore raises the staged tuple count from 204 to 52,428, and the staging area's real footprint from ~8 KiB to 52428 * row_bytes per sink per worker: ~5 MB for a 100-byte row, ~500 MB for a 10 KB row. The stage is a plain Vec, so unlike chunk bodies it is neither spillable nor pool-budgeted.

Details

The bound is real and reachable. Stage::insert ships chunk_count * chunk_capacity and stages the remainder, so the stage sits at up to chunk_capacity - 1 tuples; it is drained only by Stage::flush, which runs from consolidate_before, i.e. once per persist batch write (materialized_view_v2.rs:885). Any MV whose per-batch output exceeds 52 k updates keeps the stage at the cap between writes.

The reported size understates it by exactly the out-of-line payload: Stage::account_since passes size_of::<(D, Timestamp, Diff)>() to Accounting::stage_diff (correction_v2.rs:1886), so the stage contributes 40 bytes per record to the buffer's introspection and metrics figures regardless of row width. The comment on account_since still says "that is fine as the stage should always be small", which the new default makes false. Sink correction memory is diagnosed through exactly that figure.

Worth either capping chunk_capacity by an absolute count as well as the byte target, or deriving the ship trigger from bytes actually staged rather than from size_of of the tuple, so the flag's documented "byte size of the staging area" holds for out-of-line data.

2. LOW -- the correction buffer's spill gate is now the process-wide OR of compute's and storage's flags

src/compute/src/sink/correction_v2.rs:1596

Chunk::mint routes through chunk::try_spill_ref, whose spill_pool() returns a pool when either COMPUTE_SPILL_ENABLED or STORAGE_SPILL_ENABLED is set (src/timely-util/src/columnar/chunk.rs:184). Before this commit the buffer went through column_pager::global_pager(), gated solely on enable_column_paged_batcher_spill. Both legs are written from the same worker config aggregate (compute_state.rs:457), so turning on the storage flag enable_upsert_paged_spill now also starts spilling MV sink correction bodies on every compute replica in the process, and enable_column_paged_batcher_spill's description still reads "Allow the column-paged batcher's pager to evict chunks under memory pressure. Only meaningful when enable_column_paged_batcher = true" (src/compute-types/src/dyncfgs.rs:107).

Details

Spilling is the safer memory direction, so this is not a correctness problem, but the blast radius of both flags now includes a subsystem neither one names. Updating the two descriptions to say they also gate MV sink correction bodies would make the coupling visible to whoever rolls either flag.

@antiguru
antiguru force-pushed the moritz/correction-pool-substrate branch from 773a1d1 to 280ced7 Compare September 14, 2026 20:17
@antiguru
antiguru requested a review from a team as a code owner September 14, 2026 20:17
@antiguru

Copy link
Copy Markdown
Member Author

Both QA findings addressed; the stack is re-pushed.

1. Correct: the stage counted its fill in tuples of size_of::<(D, Timestamp, Diff)>(), 40 bytes for any Row. #38647 now sizes the stage in bytes. A DataBytes trait on the buffer's data types (Row via byte_len, DataflowErrorSer via its serialized length) reports each datum's bytes with the heap included; the stage ships once the staged bytes reach compute_correction_v2_chunk_size, and #38648's size accounting carries the same figure instead of the tuple size, so the introspection size for a correction buffer includes out-of-line row bytes. The "stage should always be small" comment is gone.

2. Both flag descriptions now say that the chunk spill gate is the OR of enable_column_paged_batcher_spill and enable_upsert_paged_spill, and that it covers the column-paged batcher and the MV sink correction buffer's chunk bodies.

🤖 Posted by Claude Code

antiguru and others added 4 commits September 14, 2026 22:21
`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 3443 ms to 1860 ms at an unchanged peak RSS. The curve is
monotone across the range, so the default is bounded only by the per-sink,
per-worker cost of the staging vector, and the `correction` bench gains a
`bench_chunk_size` sweep to show it. The flag's description now says what it
sizes.

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, the stage ships when the staged bytes reach the
flag, and its size metrics carry the same figure. `Row` answers with
`byte_len`, `DataflowErrorSer` with its serialized length.

The staging 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 grows on demand instead 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.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
`CorrectionV2::update_metrics` walked every chain and every chunk on every
operation, from four call sites including insert, so its cost grew with the
buffer: 17% of insert at 16.8M updates and rising. An `Accounting` handle,
shared by the buffer and by every structure that holds its chains, now carries
running record, size, and allocation totals, so a metrics update reads three
counters.

The totals hook the discipline the introspection logging already kept: a chain
is announced once when the buffer takes it and retired once before it is dropped
or consumed, so the totals and the logging cannot drift apart. Each chunk caches
its serialized size at construction, and each chain maintains the sum as chunks
are pushed, so no accounting step pages a chunk in.

The reported size is the logical serialized size of the bodies, resident or not.
The previous walk reported resident bytes only, which depended on what the pager
had done with each chunk since. Residency belongs to the pager, which changes it
without telling the buffer. Capacity is reported equal to size, since chunk
bodies are exactly sized when minted, and allocations count one per chunk plus
the staging vector while it holds one.

`metrics_totals_match_walk` checks the totals against a walk after inserts, a
partial drain, a since advance, and a consolidation.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
…date

Every element access in `CorrectionV2` re-decoded its chunk. `Chunk::index` was
`self.column().borrow().get(i)`, and `Column::borrow` on a serialized column
rebuilds the struct-of-arrays view from the serialized header, so a k-way merge
paid O(log k) header decodes per output update: 42.7% of an insert profile,
ahead of the merge itself at 12.2%.

Borrowing is now hoisted to one view per chunk, or per chunk pair in the two-way
merge, whose inner loop runs until either cursor crosses a chunk boundary.
`Cursor` gained `chunk_handle`, `reads_from`, and `get_with` for callers that
hold a view across cursor steps. `get` and `index` remain the documented
single-access forms. Chain iteration, chunk splitting, and the time search
index a hoisted view likewise.

Holding a view inside the iterator that `updates_before` hands across the
persist writer's `await` needs the view to be `Send`, which is a new bound on
the container. The container bounds move into a `DataContainer` helper trait
so `Data` stays readable.

Insert improves 1.6x on a 4.19M-update append workload, and 2.0x under a memory
cap, where each redundant decode also faulted the header page back in.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
`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. That gate is
the OR of `enable_column_paged_batcher_spill` and `enable_upsert_paged_spill`,
so the correction buffer now spills while either is on; both flags' descriptions
say so.

`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. `Chunk::column` takes its body
out of the pool, freeing the slot, so a chunk holds exactly one copy of its body
at any time. 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, and `mz-timely-util` a
`column_borrow` bench for the per-element borrow and encode costs the design
rests on. 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]>
@antiguru

Copy link
Copy Markdown
Member Author

Closed in favor of #38897, the same branch as a native stacked PR in this repository (stack #38898). Fork PRs cannot join GitHub stacks. The review discussion here was addressed on the branch before the move.

🤖 Posted by Claude Code

antiguru added a commit that referenced this pull request Sep 30, 2026
`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]>
antiguru added a commit that referenced this pull request Sep 30, 2026
`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.

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.

`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`. 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 stays an example: it 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.

Supersedes the fork PR #38645, which holds the earlier review
discussion.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Fable 5.1 <[email protected]>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants