From f212c869de82d006b1e99ecf791361a88482f036 Mon Sep 17 00:00:00 2001 From: Dov Alperin Date: Fri, 3 Jul 2026 02:10:55 -0400 Subject: [PATCH 1/3] Add UnloadChunk: the bulk-read capability for chunk batches UnloadChunk is the read surface for chunks whose bodies may not be resident: probe hits are copied into caller-owned staging, and no borrow of chunk contents crosses the trait boundary. Key/val/time/diff opinions stay off the trait (Staging and Probes are opaque); the one comparison the batch driver needs is delegated to the chunk via locate(), answered from resident metadata like bounds() on the cursor path. ChunkBatch gains a provided driver: gallop the chunk list by locate, open only the chunks a probe touches, and re-offer a probe left unconsumed at a chunk's last key to the next chunk, where staging's append stitches the straddling continuation (the consume-index protocol). fetch_into is the scan path. VecChunk is the worked implementation. Tests: a random-chain property test against the filter oracle, exhaustive (chunk cut x probe placement) boundary coverage including a key spanning several chunks, and a fetch round-trip. Co-Authored-By: Claude Fable 5 --- differential-dataflow/src/trace/chunk/mod.rs | 97 +++++++++++++ differential-dataflow/src/trace/chunk/vec.rs | 139 +++++++++++++++++++ 2 files changed, 236 insertions(+) diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs index 93a4d28aa..7e82e5509 100644 --- a/differential-dataflow/src/trace/chunk/mod.rs +++ b/differential-dataflow/src/trace/chunk/mod.rs @@ -24,6 +24,8 @@ //! [`ContainerChunker`](crate::trace::implementations::chunker::ContainerChunker). //! Trace *maintenance* needs only [`Chunk`]; cursor-driven *consumption* of the //! arrangement additionally asks `C` for the [`NavigableChunk`] capability. +//! Bulk consumption — copying probe hits into owned, resident staging rather than +//! navigating in place — asks for [`UnloadChunk`] instead. //! Everything else here ([`ChunkBatch`], [`ChunkMerger`], [`ChunkBatchMerger`], //! [`ChunkBatchCursor`], [`ChunkBatchBuilder`]) is machinery those aliases expand to and is //! not named directly. The [`vec`](mod@vec) module is a worked `Chunk` @@ -167,6 +169,62 @@ pub trait NavigableChunk: Chunk + Navigable: Copy; + + /// The number of probe keys. + fn probe_count(probes: Self::Probes<'_>) -> usize; + + /// Where `probes[probe_index]` falls relative to this chunk's key span: `Less` before + /// the first key, `Equal` within `[first, last]`, `Greater` past the last key. + /// + /// Resident metadata only — must never fetch a paged body. + fn locate(&self, probes: Self::Probes<'_>, probe_index: usize) -> std::cmp::Ordering; + + /// Append this chunk's updates for probes at and after `*probe_index` into `staging`, + /// advancing `*probe_index` past every probe strictly below this chunk's last key. + /// + /// A probe *equal* to the last key is extracted but not consumed: its group may + /// continue in the next chunk (see the protocol above). Any fetch of a paged + /// body is scoped to this call. + fn extract_into(&self, probes: Self::Probes<'_>, probe_index: &mut usize, staging: &mut Self::Staging); + + /// Append the whole chunk into `staging` (the scan path). + fn fetch_into(&self, staging: &mut Self::Staging); +} + /// Maximal-packing driver an implementor's [`Chunk::settle`] may delegate to. /// /// Holds a `carry` chunk under construction, grown by `combine` until it reaches @@ -255,6 +313,45 @@ impl ChunkBatch { } } +impl ChunkBatch { + /// Extract every probe hit in this batch into `staging`. + /// + /// Gallops the chunk list by [`locate`](UnloadChunk::locate) — resident metadata + /// only — and opens (via [`extract_into`](UnloadChunk::extract_into)) only the + /// chunks a probe touches. A probe left unconsumed at a chunk's last key is + /// re-offered to the next chunk, which appends the straddling continuation. + pub fn extract_into(&self, probes: C::Probes<'_>, staging: &mut C::Staging) { + use std::cmp::Ordering; + let count = C::probe_count(probes); + let chunks = &self.chunks[..]; + let (mut probe_index, mut chunk) = (0usize, 0usize); + while probe_index < count && chunk < chunks.len() { + // Whether chunk `c` lies entirely below `probes[probe_index]` (its last + // key is smaller), read from resident metadata. + let below = |c: usize| chunks[c].locate(probes, probe_index) == Ordering::Greater; + // Gallop to the first chunk not below the probe: exponential search from + // the current chunk, then binary within the final bracket. + if below(chunk) { + let (mut prev, mut step) = (chunk, 1usize); + while prev + step < chunks.len() && below(prev + step) { prev += step; step <<= 1; } + let (mut a, mut b) = (prev + 1, (prev + step).min(chunks.len())); + while a < b { let m = a + (b - a) / 2; if below(m) { a = m + 1; } else { b = m; } } + chunk = a; + } + if chunk >= chunks.len() { return; } + chunks[chunk].extract_into(probes, &mut probe_index, staging); + // Everything strictly below this chunk's last key is consumed; a probe + // equal to it was extracted but left for the next chunk (the straddle). + chunk += 1; + } + } + + /// Materialize the batch's full contents into `staging` (the scan path). + pub fn fetch_into(&self, staging: &mut C::Staging) { + for chunk in &self.chunks { chunk.fetch_into(staging); } + } +} + impl crate::trace::Navigable for ChunkBatch { type Cursor = ChunkBatchCursor; fn cursor(&self) -> Self::Cursor { diff --git a/differential-dataflow/src/trace/chunk/vec.rs b/differential-dataflow/src/trace/chunk/vec.rs index 1a6b9aff8..5f74c358f 100644 --- a/differential-dataflow/src/trace/chunk/vec.rs +++ b/differential-dataflow/src/trace/chunk/vec.rs @@ -208,6 +208,47 @@ where K: Ord+Clone+'static, V: Ord+Clone+'static, T: Lattice+Timestamp, R: Ord+S } } +impl super::UnloadChunk for VecChunk +where K: Ord+Clone+'static, V: Ord+Clone+'static, T: Lattice+Timestamp, R: Ord+Semigroup+'static { + type Staging = Vec<((K, V), T, R)>; + type Probes<'a> = &'a [K]; + + fn probe_count(probes: &[K]) -> usize { probes.len() } + + fn locate(&self, probes: &[K], probe_index: usize) -> std::cmp::Ordering { + let rows = &self.0[..]; + let probe = &probes[probe_index]; + if probe < &rows[0].0.0 { std::cmp::Ordering::Less } + else if probe > &rows[rows.len() - 1].0.0 { std::cmp::Ordering::Greater } + else { std::cmp::Ordering::Equal } + } + + fn extract_into(&self, probes: &[K], probe_index: &mut usize, staging: &mut Self::Staging) { + let rows = &self.0[..]; + let last = &rows[rows.len() - 1].0.0; + let mut pos = 0; + while *probe_index < probes.len() { + let probe = &probes[*probe_index]; + // Past this chunk's span: the probe (and everything after) is the next + // chunk's business. + if probe > last { return; } + // The probe's rows, if any: gallop to the key, then span its group. + pos = gallop(rows, pos, |u| &u.0.0 < probe); + let start = pos; + while pos < rows.len() && &rows[pos].0.0 == probe { pos += 1; } + staging.extend_from_slice(&rows[start..pos]); + // A probe equal to the last key may continue in the next chunk: + // extracted, but left unconsumed for re-offer. + if probe == last { return; } + *probe_index += 1; + } + } + + fn fetch_into(&self, staging: &mut Self::Staging) { + staging.extend_from_slice(&self.0); + } +} + impl Chunk for VecChunk where K: Ord+Clone+'static, V: Ord+Clone+'static, T: Lattice+Timestamp, R: Ord+Semigroup+'static { type Time = < as Layout>::TimeContainer as BatchContainer>::Owned; @@ -768,6 +809,104 @@ mod test { } } + // Cut a consolidated update set into a `ChunkBatch` of `sz`-row chunks, so keys + // and `(key, val)` groups straddle chunk boundaries. + fn extract_batch(updates: &[((u64, u64), u64, i64)], sz: usize) -> crate::trace::chunk::ChunkBatch> { + use crate::trace::Description; + use timely::progress::Antichain; + let chunks: Vec<_> = updates.chunks(sz).map(|c| chunk(c.to_vec())).collect(); + let desc = Description::new( + Antichain::from_elem(0u64), Antichain::from_elem(10u64), Antichain::from_elem(0u64)); + crate::trace::chunk::ChunkBatch::new(chunks, desc) + } + + // Property test: `ChunkBatch::extract_into` over random chains and random probe + // sets equals the analytic filter of the raw updates by probe membership. Tiny + // chunks force keys (with several vals and times) to straddle boundaries, + // exercising the extract/re-offer protocol; probe sets include misses and + // probes past the batch. + #[test] + fn extract_matches_filter() { + use crate::consolidation::consolidate_updates; + + let mut seed = 0x2545F4914F6CDD1Du64; + let mut rng = move || { seed ^= seed << 13; seed ^= seed >> 7; seed ^= seed << 17; seed }; + + for _ in 0..300 { + // Consolidated updates over a small key space, so groups collide and straddle. + let n = rng() as usize % 60 + 1; + let mut rows: Vec<((u64, u64), u64, i64)> = (0..n).map(|_| { + let k = rng() % 20; let v = rng() % 3; let t = rng() % 6; + let d = if rng() % 4 == 0 { -1 } else { 1 }; + ((k, v), t, d) + }).collect(); + consolidate_updates(&mut rows); + if rows.is_empty() { continue; } + let sz = rng() as usize % 5 + 1; + let batch = extract_batch(&rows, sz); + + // A sorted, deduplicated probe set over a slightly larger space (misses included). + let mut probes: Vec = (0..(rng() % 30)).map(|_| rng() % 25).collect(); + probes.sort(); + probes.dedup(); + + let mut staging = Vec::new(); + batch.extract_into(&probes[..], &mut staging); + + let want: Vec<_> = rows.iter().filter(|u| probes.binary_search(&u.0.0).is_ok()).cloned().collect(); + assert_eq!(staging, want, "chunk size {sz}\n probes={probes:?}\n rows={rows:?}"); + } + } + + // Exhaustive boundary coverage: for every chunk cut and every single-probe + // placement — below, between, at, and past the keys, including a key whose + // rows span several chunks — extraction equals the filter, and the protocol + // holds under multi-probe sets crossing the spanning key. + #[test] + fn extract_boundary_exhaustive() { + // Even keys 0..=16; key 8 carries 6 rows (distinct vals/times) so it spans + // multiple chunks at every cut size below 6. + let mut rows: Vec<((u64, u64), u64, i64)> = Vec::new(); + for k in (0..=16u64).step_by(2) { + let copies = if k == 8 { 6 } else { 1 }; + for c in 0..copies { rows.push(((k, c), c, 1)); } + } + for sz in 1..=5 { + let batch = extract_batch(&rows, sz); + // Every single probe. + for probe in 0..=18u64 { + let mut staging = Vec::new(); + batch.extract_into(&[probe][..], &mut staging); + let want: Vec<_> = rows.iter().filter(|u| u.0.0 == probe).cloned().collect(); + assert_eq!(staging, want, "sz={sz} probe={probe}"); + } + // Every contiguous probe range (hits and misses interleaved). + for lo in 0..=18u64 { + for hi in lo..=18u64 { + let probes: Vec = (lo..=hi).collect(); + let mut staging = Vec::new(); + batch.extract_into(&probes[..], &mut staging); + let want: Vec<_> = rows.iter().filter(|u| lo <= u.0.0 && u.0.0 <= hi).cloned().collect(); + assert_eq!(staging, want, "sz={sz} probes={lo}..={hi}"); + } + } + } + } + + // `fetch_into` reproduces the batch's full contents in order (the scan path). + #[test] + fn fetch_matches_contents() { + let rows: Vec<((u64, u64), u64, i64)> = + (0..40u64).map(|i| ((i / 3, i % 3), i % 5, 1)).collect(); + // `rows` is sorted and consolidated by construction. + for sz in [1, 3, 7] { + let batch = extract_batch(&rows, sz); + let mut staging = Vec::new(); + batch.fetch_into(&mut staging); + assert_eq!(staging, rows, "sz={sz}"); + } + } + // `seek_key` must land at the first key `>= target` regardless of where the cursor // starts, so the galloping hint (and its backward-seek fallback) never changes the // answer. Probe every (start, target) pair — forward and backward — against an From 5429a3f434a9e99fc449cb7cf54afa6566f34f18 Mon Sep 17 00:00:00 2001 From: Dov Alperin Date: Fri, 3 Jul 2026 02:13:13 -0400 Subject: [PATCH 2/3] Implement UnloadChunk for ColChunk; meld_keys range melds on the trie builder UpdatesBuilder gains meld_keys: meld a key-index range of a borrowed UpdatesView -- the keys with their complete val/time/diff subtrees -- by bulk column-range copies, with the same boundary-stitching semantics as meld (which now delegates to it over the full range). The view may sit over typed columns or directly over wire bytes, so one primitive serves resident and spilled sources. ColChunk's extraction gallops the probe column against the key column and melds maximal runs of consecutive hits as single range copies. On the paged arm a fetch is scoped to the call and reads a zero-copy view over the loaded bytes (spill::read) -- no trie rebuild, and the OnceCell cache stays unpopulated, unlike the cursor path's fetch-and-cache- forever. locate answers from resident metadata (ChunkMeta when paged). Tests: a filter-oracle property test over random chains for resident and paged chunks (asserting caches stay empty), a straddle fixture, a paged extraction + scan test with spill stats, and a meld_keys property test against filtered form. Co-Authored-By: Claude Fable 5 --- .../src/columnar/trace/chunk.rs | 303 ++++++++++++++++-- .../src/columnar/trace/spill.rs | 11 +- differential-dataflow/src/columnar/updates.rs | 134 +++++--- 3 files changed, 391 insertions(+), 57 deletions(-) diff --git a/differential-dataflow/src/columnar/trace/chunk.rs b/differential-dataflow/src/columnar/trace/chunk.rs index a8357ae28..2d41d6bce 100644 --- a/differential-dataflow/src/columnar/trace/chunk.rs +++ b/differential-dataflow/src/columnar/trace/chunk.rs @@ -45,7 +45,7 @@ use crate::trace::cursor::Cursor; use crate::trace::implementations::{BatchContainer, Layout, WithLayout}; use crate::columnar::layout::{ColumnarLayout, ColumnarUpdate, Coltainer}; -use crate::columnar::updates::{child_range, UpdatesBuilder, UpdatesTyped}; +use crate::columnar::updates::{child_range, UpdatesBuilder, UpdatesTyped, UpdatesView}; use crate::columnar::trie_merger; use super::spill::{self, BytesSource}; @@ -325,6 +325,116 @@ where U::Time: 'static { } } +/// Narrow a columnar ref to a shorter lifetime, so refs from different borrows — +/// a probe column and a chunk's own columns, say — can be compared (the refs are +/// lifetime-invariant). +fn rr<'b, 'a: 'b, C: Columnar>(item: columnar::Ref<'a, C>) -> columnar::Ref<'b, C> { + columnar::ContainerOf::::reborrow_ref(item) +} + +/// Extract probe hits from a trie view into `staging`, per the +/// [`UnloadChunk`](crate::trace::chunk::UnloadChunk) consume-index protocol: consume +/// probes strictly below the view's last key — melding hits, passing over misses — +/// and extract-but-don't-consume a probe equal to it (its group may continue in the +/// next chunk). A run of hits on consecutive keys melds as one bulk range copy. +/// +/// The view may sit over typed columns (a resident trie, a filled cache) or directly +/// over loaded bytes (a paged chunk, fetched for the scope of one call). +fn extract_view( + view: UpdatesView<'_, U>, + probes: as Borrow>::Borrowed<'_>, + probe_index: &mut usize, + staging: &mut UpdatesBuilder, +) { + let keys = view.keys.values; + let count = probes.len(); + let last = keys.len() - 1; + let mut key_pos = 0; + while *probe_index < count { + let probe = probes.get(*probe_index); + // Past this chunk's span: the probe (and everything after) is the next chunk's. + if rr::(probe) > rr::(keys.get(last)) { return; } + // First key `>= probe`; in range, since the probe is at most the last key. + trie_merger::gallop(keys, &mut key_pos, last + 1, |x| rr::(x) < rr::(probe)); + // Grow a run of consecutive probe-hit keys, then meld it as one range copy. + let run_start = key_pos; + while *probe_index < count && rr::(probes.get(*probe_index)) == rr::(keys.get(key_pos)) { + key_pos += 1; + if key_pos > last { + // Hit on the last key: extract the run, but leave the probe + // unconsumed — its group may continue in the next chunk. + staging.meld_keys(view, run_start..key_pos); + return; + } + *probe_index += 1; + } + if key_pos > run_start { + staging.meld_keys(view, run_start..key_pos); + } else { + // A miss, strictly below the last key: consumed silently. + *probe_index += 1; + } + } +} + +impl crate::trace::chunk::UnloadChunk for ColChunk +where U::Time: 'static { + type Staging = UpdatesBuilder; + type Probes<'a> = as Borrow>::Borrowed<'a>; + + fn probe_count(probes: Self::Probes<'_>) -> usize { probes.len() } + + fn locate(&self, probes: Self::Probes<'_>, probe_index: usize) -> std::cmp::Ordering { + use std::cmp::Ordering; + let probe = probes.get(probe_index); + let span = |first, last| { + if rr::(probe) < rr::(first) { Ordering::Less } + else if rr::(probe) > rr::(last) { Ordering::Greater } + else { Ordering::Equal } + }; + match self { + ColChunk::Resident(rc) => { + let keys = rc.view().keys.values; + span(keys.get(0), keys.get(keys.len() - 1)) + } + // Resident metadata only — no fetch. + ColChunk::Paged(p) => span(p.meta.fk.borrow().get(0), p.meta.lk.borrow().get(0)), + } + } + + fn extract_into(&self, probes: Self::Probes<'_>, probe_index: &mut usize, staging: &mut Self::Staging) { + match self { + ColChunk::Resident(rc) => extract_view(rc.view(), probes, probe_index, staging), + ColChunk::Paged(p) => match p.cache.get() { + // Already materialized (a cursor pinned it): read the cache. + Some(rc) => extract_view(rc.view(), probes, probe_index, staging), + // Fetch scoped to this call: view the loaded bytes directly — no + // trie rebuild — and drop them on return. The cache stays empty. + None => { + spill::note_fetched(); + let updates = spill::read::(&*p.source); + extract_view(updates.view(), probes, probe_index, staging); + } + }, + } + } + + fn fetch_into(&self, staging: &mut Self::Staging) { + match self { + ColChunk::Resident(rc) => staging.meld(rc), + ColChunk::Paged(p) => match p.cache.get() { + Some(rc) => staging.meld(rc), + None => { + spill::note_fetched(); + let updates = spill::read::(&*p.source); + let view = updates.view(); + staging.meld_keys(view, 0..view.keys.values.len()); + } + }, + } + } +} + impl Chunk for ColChunk where U::Time: 'static { type Time = < as Layout>::TimeContainer as BatchContainer>::Owned; @@ -546,15 +656,31 @@ fn advance_trie( #[cfg(test)] mod test { + use std::cell::RefCell; use std::collections::VecDeque; + use std::rc::Rc; use columnar::Push; - use super::{ColChunk, Chunk}; + use super::{seal_chunk, spill, ColChunk, Chunk}; + use super::spill::{BytesSource, BytesStore}; use crate::columnar::updates::UpdatesTyped; use crate::trace::chunk::merge_chains; use crate::trace::Navigable; type Upd = (u64, u64, u64, i64); + /// In-memory backing store: an arena of byte blobs. + struct MemStore(Rc>>>); + struct MemSource(Rc>>>, usize); + impl BytesStore for MemStore { + fn store(&mut self, bytes: &[u8]) -> Box { + let mut arena = self.0.borrow_mut(); + let id = arena.len(); + arena.push(bytes.to_vec()); + Box::new(MemSource(self.0.clone(), id)) + } + } + impl BytesSource for MemSource { fn load(&self) -> Vec { self.0.borrow()[self.1].clone() } } + // A sorted, consolidated columnar chunk from raw updates. fn chunk(updates: Vec) -> ColChunk { let mut u = UpdatesTyped::::default(); @@ -752,24 +878,9 @@ mod test { // exact contents (exercises the trie byte codec + the OnceCell materialization). #[test] fn settle_pages_and_round_trips() { - use std::cell::RefCell; - use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::Ordering::Relaxed; - use crate::columnar::trace::spill::{self, BytesSource, BytesStore, SpillStats}; - - // In-memory backing store: an arena of byte blobs. - struct MemStore(Rc>>>); - struct MemSource(Rc>>>, usize); - impl BytesStore for MemStore { - fn store(&mut self, bytes: &[u8]) -> Box { - let mut a = self.0.borrow_mut(); - let id = a.len(); - a.push(bytes.to_vec()); - Box::new(MemSource(self.0.clone(), id)) - } - } - impl BytesSource for MemSource { fn load(&self) -> Vec { self.0.borrow()[self.1].clone() } } + use crate::columnar::trace::spill::SpillStats; let arena = Rc::new(RefCell::new(Vec::new())); let stats = Arc::new(SpillStats::default()); @@ -792,6 +903,162 @@ mod test { spill::uninstall(); } + // A sorted, deduplicated probe column from raw keys. + fn probe_col(keys: &[u64]) -> columnar::ContainerOf { + let mut col = columnar::ContainerOf::::default(); + for k in keys { col.push(*k); } + col + } + + // Cut a consolidated update set into a `ChunkBatch` of `sz`-row chunks. + fn extract_batch(updates: &[Upd], sz: usize) -> crate::trace::chunk::ChunkBatch> { + use crate::trace::Description; + use timely::progress::Antichain; + let chunks: Vec<_> = updates.chunks(sz).map(|c| chunk(c.to_vec())).collect(); + let desc = Description::new( + Antichain::from_elem(0u64), Antichain::from_elem(10u64), Antichain::from_elem(0u64)); + crate::trace::chunk::ChunkBatch::new(chunks, desc) + } + + // Flatten extraction staging back to its update stream. + fn drain(staging: crate::columnar::updates::UpdatesBuilder) -> Vec { + staging.done().iter().map(|(k, v, t, d)| (*k, *v, *t, *d)).collect() + } + + // Property test: `ChunkBatch::extract_into` over random chains and random probe + // sets equals the analytic filter of the raw updates by probe membership — for + // resident chunks and for the same chunks paged out (extraction then reads a + // zero-copy view over the spilled bytes). Tiny chunks force keys, with several + // vals and times, to straddle boundaries, exercising the meld-stitch re-offer. + #[test] + fn extract_matches_filter() { + use columnar::Borrow; + use crate::consolidation::consolidate_updates; + + let mut seed = 0x2545F4914F6CDD1Du64; + let mut rng = move || { seed ^= seed << 13; seed ^= seed >> 7; seed ^= seed << 17; seed }; + + for paged in [false, true] { + if paged { + // Budget 0: `seal_chunk` pages everything. + let arena = Rc::new(RefCell::new(Vec::new())); + spill::install(0, Box::new(MemStore(arena)), std::sync::Arc::new(spill::SpillStats::default())); + } + for _ in 0..300 { + let n = rng() as usize % 60 + 1; + let mut rows: Vec<((u64, u64), u64, i64)> = (0..n).map(|_| { + let k = rng() % 20; let v = rng() % 3; let t = rng() % 6; + let d = if rng() % 4 == 0 { -1 } else { 1 }; + ((k, v), t, d) + }).collect(); + consolidate_updates(&mut rows); + if rows.is_empty() { continue; } + let rows: Vec = rows.into_iter().map(|((k, v), t, d)| (k, v, t, d)).collect(); + let sz = rng() as usize % 5 + 1; + let mut batch = extract_batch(&rows, sz); + if paged { + for c in batch.chunks.iter_mut() { *c = seal_chunk(std::mem::take(c)); } + assert!(batch.chunks.iter().all(|c| matches!(c, ColChunk::Paged(_)))); + } + + let mut probes: Vec = (0..(rng() % 30)).map(|_| rng() % 25).collect(); + probes.sort(); + probes.dedup(); + let col = probe_col(&probes); + + let mut staging = Default::default(); + batch.extract_into(col.borrow(), &mut staging); + let got = drain(staging); + + let want: Vec = rows.iter().filter(|u| probes.binary_search(&u.0).is_ok()).cloned().collect(); + assert_eq!(got, want, "paged={paged} chunk size {sz}\n rows={rows:?}"); + if paged { + // Extraction viewed the spilled bytes; nothing was materialized. + for c in &batch.chunks { + let ColChunk::Paged(p) = c else { unreachable!() }; + assert!(p.cache.get().is_none(), "extraction populated a chunk cache"); + } + } + } + if paged { spill::uninstall(); } + } + } + + // The extract/re-offer protocol reconstructs straddling groups: a key — and a + // `(key, val)`'s times — spanning a chunk boundary arrives in staging stitched, + // exactly as the flat filter reference has it. + #[test] + fn extract_handles_straddle() { + use columnar::Borrow; + let rows: Vec = vec![ + (0, 0, 0, 1), (1, 0, 0, 1), (1, 1, 0, 1), + (1, 1, 1, 1), (1, 2, 0, 1), + (2, 0, 0, 1), + ]; + let batch = extract_batch(&rows, 3); + for probes in [vec![1u64], vec![0, 2], vec![0, 1, 2], vec![0, 1, 2, 3]] { + let col = probe_col(&probes); + let mut staging = Default::default(); + batch.extract_into(col.borrow(), &mut staging); + let want: Vec = rows.iter().filter(|u| probes.contains(&u.0)).cloned().collect(); + assert_eq!(drain(staging), want, "probes={probes:?}"); + } + } + + // Reading paged chunks through the unload path — probe extraction and the + // fetch_into scan — reproduces exact contents while leaving every chunk's + // cache unpopulated: reads view the spilled bytes for the scope of one call + // and drop them. This is the no-pin property, versus the cursor path's + // fetch-and-cache-forever OnceCell. + #[test] + fn extract_paged_without_pinning() { + use std::sync::Arc; + use std::sync::atomic::Ordering::Relaxed; + use columnar::Borrow; + use crate::columnar::trace::spill::SpillStats; + use crate::trace::Description; + use timely::progress::Antichain; + + let arena = Rc::new(RefCell::new(Vec::new())); + let stats = Arc::new(SpillStats::default()); + spill::install(1, Box::new(MemStore(arena)), stats.clone()); // budget 1 record: page everything + + // Settle many single-update chunks into full paged chunks, then batch them. + let n = 3 * super::TARGET as u64; + let mut input: VecDeque<_> = (0..n).map(|k| chunk(vec![(k, 0, k % 5, 1)])).collect(); + let mut out = VecDeque::new(); + ColChunk::settle(&mut input, true, &mut out); + assert!(out.iter().any(|c| matches!(c, ColChunk::Paged(_))), "nothing was paged"); + let desc = Description::new( + Antichain::from_elem(0u64), Antichain::from_elem(5u64), Antichain::from_elem(0u64)); + let batch = crate::trace::chunk::ChunkBatch::new(out.into(), desc); + let unpinned = |batch: &crate::trace::chunk::ChunkBatch>| { + batch.chunks.iter().all(|c| match c { + ColChunk::Paged(p) => p.cache.get().is_none(), + ColChunk::Resident(_) => true, + }) + }; + + // Sparse probes (some misses past the key space), through the batch driver. + let probes: Vec = (0..=n + 10).step_by(97).collect(); + let col = probe_col(&probes); + let mut staging = Default::default(); + batch.extract_into(col.borrow(), &mut staging); + let want: Vec = probes.iter().filter(|&&k| k < n).map(|&k| (k, 0, k % 5, 1)).collect(); + assert_eq!(drain(staging), want); + assert!(stats.fetched_chunks.load(Relaxed) > 0, "nothing was fetched"); + assert!(unpinned(&batch), "extraction populated a chunk cache"); + + // The scan path: full contents, still nothing pinned. + let mut staging = Default::default(); + batch.fetch_into(&mut staging); + let want: Vec = (0..n).map(|k| (k, 0, k % 5, 1)).collect(); + assert_eq!(drain(staging), want); + assert!(unpinned(&batch), "fetch_into populated a chunk cache"); + + spill::uninstall(); + } + // A single `(key, val)` spanning every pushed chunk: `advance` makes no // progress until `done`, accumulating in the carry, and must still produce // the right advanced-and-consolidated result. diff --git a/differential-dataflow/src/columnar/trace/spill.rs b/differential-dataflow/src/columnar/trace/spill.rs index bdd32ed5f..5d584db7e 100644 --- a/differential-dataflow/src/columnar/trace/spill.rs +++ b/differential-dataflow/src/columnar/trace/spill.rs @@ -126,9 +126,16 @@ pub(crate) fn note_fetched() { }); } +/// Read the bytes a [`BytesSource`] produced back as an [`Updates`] whose columns +/// view the loaded bytes directly (zero-copy). Callers that only read — extraction, +/// say — should use this and drop it; [`decode`] adds the copy into typed columns. +pub(crate) fn read(source: &dyn BytesSource) -> Updates { + let bytes = timely::bytes::arc::BytesMut::from(source.load()).freeze(); + Updates::::read_from(bytes) +} + /// Reconstruct a trie from bytes a [`BytesSource`] produced (the inverse of /// `try_page`'s serialization). pub(crate) fn decode(source: &dyn BytesSource) -> UpdatesTyped { - let bytes = timely::bytes::arc::BytesMut::from(source.load()).freeze(); - Updates::::read_from(bytes).into_typed() + read::(source).into_typed() } diff --git a/differential-dataflow/src/columnar/updates.rs b/differential-dataflow/src/columnar/updates.rs index cbc4244fe..64151da6e 100644 --- a/differential-dataflow/src/columnar/updates.rs +++ b/differential-dataflow/src/columnar/updates.rs @@ -599,6 +599,10 @@ pub struct UpdatesBuilder { updates: UpdatesTyped, } +impl Default for UpdatesBuilder { + fn default() -> Self { Self { updates: UpdatesTyped::default() } } +} + impl UpdatesBuilder { /// Construct a new builder from consolidated, sealed updates. /// @@ -622,39 +626,57 @@ impl UpdatesBuilder { /// (continue the current group), but times must be strictly increasing /// within the same `(key, val)`. pub fn meld(&mut self, chunk: &UpdatesTyped) { + use columnar::Len; + self.meld_keys(chunk.view(), 0..Len::len(&chunk.keys.values)); + } + + /// Meld `view[key_range]` — the keys in the range with their complete val, time, + /// and diff subtrees — into this builder, by bulk column-range copies. + /// + /// The same precondition as [`meld`](Self::meld), applied to the range: its first + /// `(key, val, time)` must be strictly greater than the builder's last. The key + /// and val may equal the builder's open key/val (continuing that group), as long + /// as the time is strictly greater. `meld` is this over a chunk's full key range; + /// extraction melds the probe-hit ranges of a chunk it never fully copies. + pub fn meld_keys(&mut self, view: UpdatesView<'_, U>, key_range: std::ops::Range) { use columnar::{Borrow, Index, Len}; - if chunk.len() == 0 { return; } + // Narrow both sides of a key/val/time comparison to a common lifetime + // (columnar refs are invariant). + fn rr<'b, 'a: 'b, C: Columnar>(item: columnar::Ref<'a, C>) -> columnar::Ref<'b, C> { + columnar::ContainerOf::::reborrow_ref(item) + } + + if key_range.is_empty() { return; } + let val_range = view.vals_bounds(key_range.clone()); + let time_range = view.times_bounds(val_range.clone()); - // Empty builder: clone the chunk and unseal it. + // Empty builder: bulk-copy the range, leaving the trailing groups open. if Len::len(&self.updates.keys.values) == 0 { - self.updates = chunk.clone(); - self.updates.keys.bounds.pop(); + self.updates.keys.values.extend_from_self(view.keys.values, key_range.clone()); + self.updates.vals.extend_from_self(view.vals, key_range); self.updates.vals.bounds.pop(); + self.updates.times.extend_from_self(view.times, val_range.clone()); self.updates.times.bounds.pop(); + self.updates.diffs.extend_from_self(view.diffs, time_range); return; } // Pre-compute boundary comparisons before mutating. let keys_match = { let skb = self.updates.keys.values.borrow(); - let ckb = chunk.keys.values.borrow(); - skb.get(Len::len(&skb) - 1) == ckb.get(0) + rr::(skb.get(Len::len(&skb) - 1)) == rr::(view.keys.values.get(key_range.start)) }; + + // Child ranges for the first element at each level of the range. + let first_key_vals = child_range(view.vals.bounds, key_range.start); + let first_val_times = child_range(view.times.bounds, first_key_vals.start); + let vals_match = keys_match && { let svb = self.updates.vals.values.borrow(); - let cvb = chunk.vals.values.borrow(); - svb.get(Len::len(&svb) - 1) == cvb.get(0) + rr::(svb.get(Len::len(&svb) - 1)) == rr::(view.vals.values.get(first_key_vals.start)) }; - let chunk_num_keys = Len::len(&chunk.keys.values); - let chunk_num_vals = Len::len(&chunk.vals.values); - let chunk_num_times = Len::len(&chunk.times.values); - - // Child ranges for the first element at each level of the chunk. - let first_key_vals = child_range(chunk.vals.borrow().bounds, 0); - let first_val_times = child_range(chunk.times.borrow().bounds, 0); - // There is a first position where coordinates disagree. // Strictly beyond that position: seal bounds, extend lists, re-open the last bound. // At that position: meld the first list, extend subsequent lists, re-open. @@ -663,12 +685,12 @@ impl UpdatesBuilder { // --- Keys --- if keys_match { // Skip the duplicate first key; add remaining keys. - if chunk_num_keys > 1 { - self.updates.keys.values.extend_from_self(chunk.keys.values.borrow(), 1..chunk_num_keys); + if key_range.len() > 1 { + self.updates.keys.values.extend_from_self(view.keys.values, (key_range.start + 1)..key_range.end); } } else { // All keys are new. - self.updates.keys.values.extend_from_self(chunk.keys.values.borrow(), 0..chunk_num_keys); + self.updates.keys.values.extend_from_self(view.keys.values, key_range.clone()); differ = true; } @@ -676,7 +698,7 @@ impl UpdatesBuilder { if differ { // Keys differed: seal open val group, extend all val lists, unseal last. self.updates.vals.bounds.push(Len::len(&self.updates.vals.values) as u64); - self.updates.vals.extend_from_self(chunk.vals.borrow(), 0..chunk_num_keys); + self.updates.vals.extend_from_self(view.vals, key_range.clone()); self.updates.vals.bounds.pop(); } else { // Keys matched: meld vals for the shared key. @@ -684,22 +706,19 @@ impl UpdatesBuilder { // Skip the duplicate first val; add remaining vals from the first key's list. if first_key_vals.len() > 1 { self.updates.vals.values.extend_from_self( - chunk.vals.values.borrow(), + view.vals.values, (first_key_vals.start + 1)..first_key_vals.end, ); } } else { // First val differs: add all vals from the first key's list. - self.updates.vals.values.extend_from_self( - chunk.vals.values.borrow(), - first_key_vals.clone(), - ); + self.updates.vals.values.extend_from_self(view.vals.values, first_key_vals.clone()); differ = true; } // Seal the matched key's val group, extend remaining keys' val lists, unseal. - if chunk_num_keys > 1 { + if key_range.len() > 1 { self.updates.vals.bounds.push(Len::len(&self.updates.vals.values) as u64); - self.updates.vals.extend_from_self(chunk.vals.borrow(), 1..chunk_num_keys); + self.updates.vals.extend_from_self(view.vals, (key_range.start + 1)..key_range.end); self.updates.vals.bounds.pop(); } } @@ -708,26 +727,22 @@ impl UpdatesBuilder { if differ { // Seal open time group, extend all time lists, unseal last. self.updates.times.bounds.push(Len::len(&self.updates.times.values) as u64); - self.updates.times.extend_from_self(chunk.times.borrow(), 0..chunk_num_vals); + self.updates.times.extend_from_self(view.times, val_range.clone()); self.updates.times.bounds.pop(); } else { // Keys and vals matched. Times must be strictly greater (precondition), // so we always set differ = true here. debug_assert!({ let stb = self.updates.times.values.borrow(); - let ctb = chunk.times.values.borrow(); - stb.get(Len::len(&stb) - 1) != ctb.get(0) + rr::(stb.get(Len::len(&stb) - 1)) != rr::(view.times.values.get(first_val_times.start)) }, "meld: duplicate time within same (key, val)"); // Add times from the first val's time list into the open group. - self.updates.times.values.extend_from_self( - chunk.times.values.borrow(), - first_val_times.clone(), - ); + self.updates.times.values.extend_from_self(view.times.values, first_val_times.clone()); differ = true; // Seal the matched val's time group, extend remaining vals' time lists, unseal. - if chunk_num_vals > 1 { + if val_range.len() > 1 { self.updates.times.bounds.push(Len::len(&self.updates.times.values) as u64); - self.updates.times.extend_from_self(chunk.times.borrow(), 1..chunk_num_vals); + self.updates.times.extend_from_self(view.times, (val_range.start + 1)..val_range.end); self.updates.times.bounds.pop(); } } @@ -737,7 +752,7 @@ impl UpdatesBuilder { // times are strictly increasing for the same (key, val), differ is // always true by this point — just extend all diff lists. debug_assert!(differ); - self.updates.diffs.extend_from_self(chunk.diffs.borrow(), 0..chunk_num_times); + self.updates.diffs.extend_from_self(view.diffs, time_range); } /// Seal all open bounds and return the completed `UpdatesTyped`. @@ -855,6 +870,51 @@ mod tests { assert_eq!(collect(&updates.consolidate()), vec![(1, 20, 100, 5)]); } + // Property test: melding arbitrary (sorted, disjoint) key-index ranges of a + // consolidated trie must equal forming the trie of just those keys' updates. + // Covers empty-builder starts, key-adjacent ranges (continuing groups), and + // gaps — the shapes extraction produces. + #[test] + fn meld_keys_matches_filtered_form() { + use columnar::Len; + + let mut seed = 0x9E3779B97F4A7C15u64; + let mut rng = move || { seed ^= seed << 13; seed ^= seed >> 7; seed ^= seed << 17; seed }; + + for _ in 0..300 { + // A consolidated trie over a small space, so keys carry several vals and times. + let n = rng() as usize % 60 + 1; + let mut source = UpdatesTyped::::default(); + for _ in 0..n { + let (k, v, t) = (rng() % 8, rng() % 3, rng() % 4); + let d: i64 = if rng() % 4 == 0 { -1 } else { 1 }; + source.push((&k, &v, &t, &d)); + } + let source = source.consolidate(); + let num_keys = Len::len(&source.keys.values); + if num_keys == 0 { continue; } + + // A random subset of key indices, grouped into maximal consecutive ranges. + let selected: Vec = (0..num_keys).filter(|_| rng() % 2 == 0).collect(); + let mut builder = UpdatesBuilder::::default(); + let mut i = 0; + while i < selected.len() { + let start = selected[i]; + let mut end = start + 1; + while i + 1 < selected.len() && selected[i + 1] == end { end += 1; i += 1; } + builder.meld_keys(source.view(), start..end); + i += 1; + } + + // Reference: the trie formed from the selected keys' updates alone. + let keys: Vec = selected.iter() + .map(|&i| *columnar::Index::get(&columnar::Borrow::borrow(&source.keys.values), i)) + .collect(); + let want: Vec<_> = collect(&source).into_iter().filter(|u| keys.contains(&u.0)).collect(); + assert_eq!(collect(&builder.done()), want, "selected key indices: {selected:?}"); + } + } + #[test] fn test_interleaved_cancellations() { let mut updates = UpdatesTyped::::default(); From b27ad576c5634afc345b446f81ef686d403b9514 Mon Sep 17 00:00:00 2001 From: Dov Alperin Date: Fri, 3 Jul 2026 02:21:32 -0400 Subject: [PATCH 3/3] chunk_bench: probe extraction vs cursor navigation A read-side section for the UnloadChunk path: both paths consume the probed updates of a columnar ChunkBatch into owned staging -- the straddle cursor by per-probe seeks and owned copies, extraction by one extract_into over trie staging. Shapes: dense String-keyed probes (the design doc's headline case), sparse (1%) u64 probes, dense u64 probes, and fat 4x256B values; the first two also run against paged batches read cold, where the cursor path decodes-and-caches and extraction views the spilled bytes. At 1M updates on an M-series laptop: dense String probes 260ms cursor vs 12.9ms extraction (20x; the design doc's 21ms-class admission), sparse probes at 2x better than cursor parity, fat values 3.5x with extract_into at the pure-meld copy floor (fetch_into parity, so the probe machinery costs nothing). Paged cold reads widen every gap. Co-Authored-By: Claude Fable 5 --- differential-dataflow/benches/chunk_bench.rs | 189 +++++++++++++++++++ 1 file changed, 189 insertions(+) diff --git a/differential-dataflow/benches/chunk_bench.rs b/differential-dataflow/benches/chunk_bench.rs index 33e34f9cd..9eee0ea3e 100644 --- a/differential-dataflow/benches/chunk_bench.rs +++ b/differential-dataflow/benches/chunk_bench.rs @@ -187,6 +187,195 @@ fn main() { let b: Vec<_> = u.iter().filter(|(_, t, _)| *t >= half / pairs).cloned().collect(); bench_valbearing("D: few (k,v), many times, time-disjoint merge (LSM append)", u, a, b); } + + // Shape X: read-side unloading — probe extraction vs cursor navigation. + extraction::run(n); +} + +/// Probe-extraction (`UnloadChunk`) vs cursor navigation over a `ChunkBatch` of +/// columnar chunks: both consume the probed updates into owned staging — the +/// cursor by per-probe seeks and owned copies, extraction by bulk column-range +/// melds. Resident and paged (cold spilled bytes) variants. +mod extraction { + use std::cell::RefCell; + use std::collections::VecDeque; + use std::rc::Rc; + use std::time::Instant; + + use columnar::{Borrow, Columnar, ContainerOf, Index, Len, Push}; + use timely::progress::Antichain; + + use differential_dataflow::columnar::layout::{ColumnarUpdate, Coltainer}; + use differential_dataflow::columnar::trace::ColChunk; + use differential_dataflow::columnar::trace::spill::{self, BytesSource, BytesStore, SpillStats}; + use differential_dataflow::columnar::updates::UpdatesBuilder; + use differential_dataflow::trace::chunk::{Chunk, ChunkBatch}; + use differential_dataflow::trace::cursor::Cursor; + use differential_dataflow::trace::implementations::BatchContainer; + use differential_dataflow::trace::{Description, Navigable}; + + use super::{build_chain, ms}; + + /// In-memory backing store: an arena of byte blobs. + struct MemStore(Rc>>>); + struct MemSource(Rc>>>, usize); + impl BytesStore for MemStore { + fn store(&mut self, bytes: &[u8]) -> Box { + let mut arena = self.0.borrow_mut(); + let id = arena.len(); + arena.push(bytes.to_vec()); + Box::new(MemSource(self.0.clone(), id)) + } + } + impl BytesSource for MemSource { fn load(&self) -> Vec { self.0.borrow()[self.1].clone() } } + + fn batch_of>(chunks: Vec>) -> ChunkBatch> { + let desc = Description::new( + Antichain::from_elem(0u64), Antichain::from_elem(1u64), Antichain::from_elem(0u64)); + ChunkBatch::new(chunks, desc) + } + + /// Page every chunk of `chain` out through `settle` (budget 0), leaving the + /// spiller installed so reads count fetches. + fn page_out>(chain: Vec>) -> Vec> { + let arena = Rc::new(RefCell::new(Vec::new())); + spill::install(0, Box::new(MemStore(arena)), std::sync::Arc::new(SpillStats::default())); + let mut input: VecDeque<_> = chain.into(); + let mut out = VecDeque::new(); + ColChunk::settle(&mut input, true, &mut out); + out.into() + } + + /// The cursor path: per-probe `seek_key` on the straddle cursor, copying the + /// hit's key, vals, times, and diffs out as owned values. Returns updates staged. + fn probe_cursor>(batch: &ChunkBatch>, probes: &ContainerOf) -> usize { + let mut cursor = batch.cursor(); + let mut staged: Vec<(U::Key, U::Val, U::Time, U::Diff)> = Vec::new(); + let pb = probes.borrow(); + for i in 0..pb.len() { + let probe = pb.get(i); + cursor.seek_key(batch, probe); + let Some(key) = cursor.get_key(batch) else { continue }; + if as BatchContainer>::reborrow(key) + != as BatchContainer>::reborrow(probe) { continue; } + let key = ::into_owned(key); + while let Some(val) = cursor.get_val(batch) { + let val = ::into_owned(val); + let k = key.clone(); + cursor.map_times(batch, |t, d| staged.push(( + k.clone(), val.clone(), + ::into_owned(t), + ::into_owned(d), + ))); + cursor.step_val(batch); + } + } + staged.len() + } + + /// The extraction path: one `extract_into` over the batch into trie staging. + fn probe_extract>(batch: &ChunkBatch>, probes: &ContainerOf) -> usize { + let mut staging = UpdatesBuilder::::default(); + batch.extract_into(probes.borrow(), &mut staging); + staging.done().len() + } + + /// Best-of-`reps` wall clock; returns (ms, result) and asserts result stability. + fn best(reps: usize, mut f: impl FnMut() -> usize) -> (f64, usize) { + let (mut best, mut out) = (f64::MAX, 0); + for r in 0..reps { + let t = Instant::now(); + let got = std::hint::black_box(f()); + best = best.min(ms(t)); + if r > 0 { assert_eq!(got, out); } + out = got; + } + (best, out) + } + + fn row(label: &str, hits: usize, cursor_ms: f64, extract_ms: f64) { + println!(" {label:<38} {hits:>9} {cursor_ms:>10.1} {extract_ms:>11.1} {:>10.2}x", cursor_ms / extract_ms); + } + + /// One shape: resident best-of-3 both paths, then (optionally) a paged batch + /// measured cold — a single pass each, on separately paged batches, since the + /// cursor path materializes chunk caches as it reads (extraction does not). + fn shape>(label: &str, updates: Vec<((U::Key, U::Val), u64, i64)>, probes: &ContainerOf, paged: bool) + where + U: ColumnarUpdate, + ColChunk: timely::container::SizableContainer + + differential_dataflow::consolidation::Consolidate + + timely::container::PushInto<((U::Key, U::Val), u64, i64)>, + U::Key: Clone, U::Val: Clone, + { + let chain = build_chain::, _>(updates.clone()); + let batch = batch_of(chain); + let (cursor_ms, cursor_hits) = best(3, || probe_cursor(&batch, probes)); + let (extract_ms, extract_hits) = best(3, || probe_extract(&batch, probes)); + assert_eq!(cursor_hits, extract_hits, "paths disagree on {label}"); + row(label, extract_hits, cursor_ms, extract_ms); + + if paged { + // Cold cursor pass: first touch decodes and caches every chunk it opens. + let batch = batch_of(page_out(build_chain::, _>(updates.clone()))); + let t = Instant::now(); + let cursor_hits = std::hint::black_box(probe_cursor(&batch, probes)); + let cursor_ms = ms(t); + spill::uninstall(); + // Extraction never populates caches, so every pass is cold; best-of-3. + let batch = batch_of(page_out(build_chain::, _>(updates))); + let (extract_ms, extract_hits) = best(3, || probe_extract(&batch, probes)); + spill::uninstall(); + assert_eq!(cursor_hits, extract_hits, "paged paths disagree on {label}"); + row(&format!("{label}, paged (cold)"), extract_hits, cursor_ms, extract_ms); + } + } + + pub fn run(n: usize) { + println!("\n== probe extraction (UnloadChunk) vs cursor navigation, col trie =="); + println!(" {:<38} {:>9} {:>10} {:>11} {:>11}", "shape", "hits", "cursor ms", "extract ms", "cur/ext"); + + // Dense String keys: every key probed — the design doc's headline case. + { + let key = |k: usize| format!("key-{k:012}"); + let updates: Vec<((String, ()), u64, i64)> = (0..n).map(|k| ((key(k), ()), 0, 1)).collect(); + let mut probes = ContainerOf::::default(); + for k in 0..n { probes.push(&key(k)); } + shape::<(String, (), u64, i64)>("dense probes, String keys", updates, &probes, true); + } + + // Sparse (1%) u64 probes: navigation's best case — extraction wants parity. + { + let updates: Vec<((u64, ()), u64, i64)> = (0..n as u64).map(|k| ((k, ()), 0, 1)).collect(); + let mut probes = ContainerOf::::default(); + for k in (0..n as u64).step_by(100) { probes.push(k); } + shape::<(u64, (), u64, i64)>("sparse probes (1%), u64 keys", updates, &probes, true); + } + + // Dense u64 probes: the pure navigate-vs-copy comparison, no string compares. + { + let updates: Vec<((u64, ()), u64, i64)> = (0..n as u64).map(|k| ((k, ()), 0, 1)).collect(); + let mut probes = ContainerOf::::default(); + for k in 0..n as u64 { probes.push(k); } + shape::<(u64, (), u64, i64)>("dense probes, u64 keys", updates, &probes, false); + } + + // Fat values: 4 x 256B string vals per key, every key probed — value copies + // dominate, and staging should run at memory bandwidth. + { + let keys = (n / 16).max(1) as u64; + let fat = |k: u64, v: u64| { + let mut s = format!("val-{k:012}-{v:02}-"); + s.push_str(&"x".repeat(256 - s.len())); + s + }; + let updates: Vec<((u64, String), u64, i64)> = + (0..keys).flat_map(|k| (0..4).map(move |v| ((k, fat(k, v)), 0, 1))).collect(); + let mut probes = ContainerOf::::default(); + for k in 0..keys { probes.push(k); } + shape::<(u64, String, u64, i64)>("dense probes, fat vals (4 x 256B)", updates, &probes, false); + } + } } /// Shape B/D need a `u64` val type (not `()`), so they get their own dispatch,