Repository navigation
Conversation
antiguru
left a comment
There was a problem hiding this comment.
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?
| #[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))) | ||
| } | ||
|
|
There was a problem hiding this comment.
Between comment and function.
There was a problem hiding this comment.
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
| // Columnation regions are not Stacked Borrows compliant: later pushes invalidate the | ||
| // provenance of previously stored items under Miri. |
There was a problem hiding this comment.
Fine to ignore miri here, but we'ren't using columnation here anymore.
There was a problem hiding this comment.
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) { |
There was a problem hiding this comment.
Do we ever pass a non-0 depth? We could for deeper chains in the correction structure.
There was a problem hiding this comment.
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
| // 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. |
There was a problem hiding this comment.
"Serialize into a vec to leave allocation in place, and avoid shipping spare capacity." Something like this?
There was a problem hiding this comment.
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
| /// 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. |
There was a problem hiding this comment.
Let's avoid the measurement results, and just say it avoids the per-element borrowing cost.
There was a problem hiding this comment.
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
| /// 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. |
There was a problem hiding this comment.
The 2MiB number is a property of the column builder, so just say we defer column formation to it.
There was a problem hiding this comment.
Reworded: the builder mints as soon as the column reports itself at capacity (Column::at_capacity), without naming the size.
🤖 Posted by Claude Code
| + 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 |
There was a problem hiding this comment.
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
d11c7c9 to
773a1d1
Compare
|
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 The 🤖 Posted by Claude Code |
QA LLM Review1. MEDIUM --
|
773a1d1 to
280ced7
Compare
|
Both QA findings addressed; the stack is re-pushed. 1. Correct: the stage counted its fill in tuples of 2. Both flag descriptions now say that the chunk spill gate is the OR of 🤖 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 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]>
280ced7 to
47662fb
Compare
|
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 |
`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]>
`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]>
Stacked on #38649. Review the last commit; the diff shows #38647, #38648, and #38649 as well until they merge.
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.ChunkBuilderno longer wrapsColumnBuilder: it owns one column, mints as soon as the column reports itself at capacity, and returns the minted chunk frompush.Chunk::columntakes 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
correctionbench gains a spill sweep, andmz-timely-utilacolumn_borrowbench for the per-element borrow and encode costs the design rests on. 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.🤖 Generated with Claude Code