timely-util: ColumnChunk, a pool-spillable Chunk implementation - #37955
Draft
DAlperin wants to merge 3 commits into
Draft
timely-util: ColumnChunk, a pool-spillable Chunk implementation#37955DAlperin wants to merge 3 commits into
DAlperin wants to merge 3 commits into
Conversation
DAlperin
force-pushed
the
dov/column-chunk
branch
from
July 30, 2026 15:40
957ea65 to
9736cb9
Compare
DAlperin
commented
Jul 30, 2026
Comment on lines
+306
to
+325
| if let Some(pool) = spill_pool() { | ||
| let len_bytes = column.length_in_bytes(); | ||
| if len_bytes >= SPILL_MIN_BYTES { | ||
| let view = column.borrow(); | ||
| let records = view.len(); | ||
| let mut first = D::Container::default(); | ||
| let mut last = D::Container::default(); | ||
| first.push(view.0.get(0)); | ||
| last.push(view.0.get(records - 1)); | ||
| let handle = spill_column(column, &pool, len_bytes, ChunkHints { depth }); | ||
| return ColumnChunk::Spilled(Rc::new(SpilledBody { | ||
| records, | ||
| first, | ||
| last, | ||
| depth, | ||
| handle, | ||
| })); | ||
| } | ||
| } | ||
| ColumnChunk::Resident(Rc::new(column), depth) |
Member
Author
There was a problem hiding this comment.
an adversarial caller could do a bad job merging and leave lots of < SPILL_MIN_BYTES chunks around, but probably fine to just... trust they won't do that
Comment on lines
+501
to
+505
| let [col_a, col_b] = &mut cols; | ||
| for (col, pos, depth, spilled, queue) in [ | ||
| (col_a, positions[0], depths[0], &mut spill_a, in1), | ||
| (col_b, positions[1], depths[1], &mut spill_b, in2), | ||
| ] { |
Member
Author
There was a problem hiding this comment.
better names or a comment
| let mut keep_col: Column<(D, T, R)> = Column::default(); | ||
| let mut ship_col: Column<(D, T, R)> = Column::default(); | ||
| while pos < len { | ||
| col.extract(&mut pos, frontier, residual, &mut keep_col, &mut ship_col); |
Member
Author
There was a problem hiding this comment.
follow up note (not for this PR) we should rewrite the underlying extract to be two pass, just the T column first then per range copies.
| /// copied once on arrival, keeping the run linear. Advancing is | ||
| /// lattice-monotone but not order-monotone, so each group's advanced | ||
| /// times are re-sorted before adjacent equal times fold. | ||
| fn advance( |
Member
Author
There was a problem hiding this comment.
Possible follow-up: since the output leaves are addressed independently, a group that folds nothing (no time collisions, no zeroed diffs) could bulk extend_from_self the D leaf over the whole group range and push only the advanced times/diffs per record, instead of re-pushing group_d per survivor. Singleton groups (the common case for mostly-unique D) could skip the scratch/sort round-trip entirely.
The ship signal was a 10% window below each 2 MiB boundary, so a single record wider than the window stepped clear over it and the container kept growing: merge and extract cuts, the builder ship point, and Column::at_capacity could all let a chunk outgrow the buffer pool's largest size class, past which a spilled body degrades to permanently resident. The threshold is now monotone at 10% under 2 MiB, which behaves identically for record-at-a-time growth and cuts wide-record chunks at the first check. The builder's inline copy of the check now calls the shared predicate.
DAlperin
force-pushed
the
dov/column-chunk
branch
from
July 30, 2026 19:00
9736cb9 to
a2170a0
Compare
DAlperin
force-pushed
the
dov/column-chunk
branch
from
July 30, 2026 19:37
a2170a0 to
c0774e1
Compare
DAlperin
force-pushed
the
dov/column-chunk
branch
from
July 30, 2026 19:52
c0774e1 to
08396b3
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
Part of the buffer-managed dataflow state design (
doc/developer/design/20260610_buffer_managed_state.md). The pool merged in #37718 and its configuration in #37719. This PR adds the chunk type that spills through it. Stacked on #37954 (the UnloadChunk trait); follow-ups adopt it in storage's upsert stash and compute's arrange sites, staged with CI on #37805.Part of CPU-184.
Description
Three commits, reviewable separately.
Monotone columnar ship threshold. The ship signal was a 10% window below each 2 MiB boundary. A single record wider than the window steps clear over it, the signal un-fires, and a chunk can grow past the pool's largest size class (8 MiB), where a spilled body degrades to permanently resident. The threshold is now monotone at 10% under 2 MiB. This touches live paths (
ColumnBuilder's ship point, merge/extract cuts,Column::at_capacity): rows under ~200 KiB serialized behave identically, wider rows now ship at the first boundary instead of drifting.ColumnChunk. Differential's
ChunkoverColumn-shaped updates: sorted, consolidated(D, T, R)runs with the merge/extract/advance/settle transducers, plus the UnloadChunk implementation for(K, V)data (locate from key fences, gallop-based extraction). Grading is by serialized bytes rather than the record-countTARGET, since record count does not bound bytes for variable-width data.Pool spilling and depth hints. A chunk is Resident (
Rc-sharedColumn) or Spilled (serialized body in the process pool, with record count and first/last fences resident).settleis the commit point: bodies at or above 64 KiB spill when the compute or storage gate is set and a pool is configured. Reads are copy-out and call-scoped, which is what lets the pool evict with no reader accounting. Each chunk carries a generational depth (fresh 0, merge output one past its deepest input) that becomes the pool's eviction-band hint, so repeatedly merged, colder data evicts first. The UnloadChunk probe path reads spilled bodies for the scope of one call and deliberately does not re-admit them.Nothing in production sets the spill gates yet. The storage and compute wiring comes in the follow-up PRs, so apart from commit 1 this is inert until then.
Verification
Property tests drive the trait methods the way the differential harness does and compare against brute-force references, in resident and force-spilled variants (batcher round trip, seal partitioning at intermediate frontiers, advance, extraction with straddled keys). Deterministic large-data tests cover the cut paths the proptests cannot reach: advance's multi-chunk cut, the giant-group carry, extract's mid-chunk cut on both sides. The threshold change carries a regression test that fails against the windowed predicate. The spill-gate matrix (compute/storage OR, no clobber) runs against a real installed pool, and the spill round trip is asserted byte-identical, including re-spilling an already-serialized body. The full stack runs CI on #37805.