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, 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(); 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