From 7c7bfc96228dc0475f337a33dfcd079e4e9855ee Mon Sep 17 00:00:00 2001 From: Dov Alperin Date: Tue, 8 Sep 2026 15:33:38 -0400 Subject: [PATCH 01/31] timely-util: bound serialized column chunks and merge allocations --- src/timely-util/src/columnar/batcher.rs | 94 ++++++--------- src/timely-util/src/columnar/chunk.rs | 152 ++++++++++++++++++++++-- 2 files changed, 177 insertions(+), 69 deletions(-) diff --git a/src/timely-util/src/columnar/batcher.rs b/src/timely-util/src/columnar/batcher.rs index d459ea2efebdc..116e911fb4f43 100644 --- a/src/timely-util/src/columnar/batcher.rs +++ b/src/timely-util/src/columnar/batcher.rs @@ -21,7 +21,7 @@ use std::marker::PhantomData; use crate::columnation::ColumnationStack; use columnar::Container as _; use columnar::Push as _; -use columnar::{Clear, Columnar, Index, Len}; +use columnar::{Borrow, Clear, Columnar, Index, Len}; use columnation::Columnation; use differential_dataflow::difference::Semigroup; use differential_dataflow::trace::implementations::merge_batcher::Merger; @@ -29,6 +29,7 @@ use timely::Accountable; use timely::Container; use timely::PartialOrder; use timely::container::{ContainerBuilder, PushInto, SizableContainer}; +use timely::dataflow::channels::ContainerBytes; use timely::progress::frontier::{Antichain, AntichainRef}; use crate::columnar::Column; @@ -382,22 +383,13 @@ where // every record path pushes exactly one element to each. let (sd, st, sr) = self_c; - // Pre-size each output leaf for the worst-case merge - // (no consolidation): `len(left) + len(right)` records. - // `reserve_for` walks each input's `as_bytes`, which is - // accurate for variable-length leaves (where reserving a - // record count wouldn't size the byte buffer correctly). - // - // Gated by record count: above a few hundred thousand - // records the input bound over-reserves any time - // consolidation is heavy, and the framework's outer - // ship-threshold check yields us before we'd use the - // headroom. For inputs past that point, geometric grow - // is bounded by 2× the actual output and avoids - // committing pages we'd never touch. - const RESERVE_RECORD_THRESHOLD: usize = 1_000_000; - if upper_l + upper_r <= RESERVE_RECORD_THRESHOLD { - use columnar::Container as _; + // Reserving by record count can allocate an entire wide input + // for every small output chunk. Reserve only when both inputs + // fit within two ship-sized chunks. + let input_bytes = left[0] + .length_in_bytes() + .saturating_add(right[0].length_in_bytes()); + if input_bytes <= 2 * crate::columnar::SHIP_WORDS * 8 { let inputs = [left_borrow, right_borrow]; sd.reserve_for(inputs.iter().map(|b| b.0)); st.reserve_for(inputs.iter().map(|b| b.1)); @@ -406,33 +398,14 @@ where let mut stash = R::default(); - // Mid-merge ship-threshold check, matching the heuristic - // used by `Column::at_capacity` and `ColumnBuilder`. The - // tuple `(sd.borrow(), st.borrow(), sr.borrow())` chains - // its leaves' `as_bytes` iterators, so passing it to - // `at_serialized_capacity` reuses the canonical - // `indexed::length_in_words` formula without needing the - // parent borrow we destructured. - // - // The check walks every leaf slice once per call, which - // is non-trivial on variable-length leaves; the caller - // runs it every `THRESHOLD_PERIOD_MASK + 1` iterations - // rather than per-iter. The ship threshold is ~65 K - // records, so overshooting by ~1 K records before the - // check fires has no practical impact — the framework's - // outer `at_capacity` check sees the oversize chunk and - // ships it regardless. - let at_ship_threshold = - |sd: &D::Container, st: &T::Container, sr: &R::Container| { - use columnar::Borrow as _; - crate::columnar::at_serialized_capacity(&( - sd.borrow(), - st.borrow(), - sr.borrow(), - )) - }; - const THRESHOLD_PERIOD_MASK: u32 = 1023; - let mut iter: u32 = 0; + // Spend at most the ship threshold's headroom between size + // checks at the input's average width. This is a cadence hint, + // not a hard bound for heterogeneous rows. Chunk settlement + // splits any oversized result at record boundaries. + let average_bytes = input_bytes / (upper_l + upper_r).max(1); + let check_records = + (crate::columnar::SHIP_WORDS * 8 / 10 / average_bytes.max(1)).clamp(1, 1024); + let mut next_check = sd.len() + check_records; let mut yielded = false; while left_pos[0] < upper_l && right_pos[0] < upper_r { @@ -461,9 +434,11 @@ where && (l_d.get(left_pos[0]), l_t.get(left_pos[0])) < (d2, t2) { let start = left_pos[0]; - gallop(upper_l, &mut left_pos[0], |i| { - (l_d.get(i), l_t.get(i)) < (d2, t2) - }); + gallop( + upper_l.min(left_pos[0] + next_check.saturating_sub(sd.len())), + &mut left_pos[0], + |i| (l_d.get(i), l_t.get(i)) < (d2, t2), + ); // Per-leaf bulk copy of the run: each call // resolves to an `extend_from_slice` on its // leaf (recursively for nested leaves). @@ -482,9 +457,11 @@ where && (r_d.get(right_pos[0]), r_t.get(right_pos[0])) < (d1, t1) { let start = right_pos[0]; - gallop(upper_r, &mut right_pos[0], |i| { - (r_d.get(i), r_t.get(i)) < (d1, t1) - }); + gallop( + upper_r.min(right_pos[0] + next_check.saturating_sub(sd.len())), + &mut right_pos[0], + |i| (r_d.get(i), r_t.get(i)) < (d1, t1), + ); sd.extend_from_self(r_d, start..right_pos[0]); st.extend_from_self(r_t, start..right_pos[0]); sr.extend_from_self(r_r, start..right_pos[0]); @@ -505,12 +482,16 @@ where } } - // Amortized ship-threshold check; see comment above - // `at_ship_threshold` for rationale. - iter = iter.wrapping_add(1); - if iter & THRESHOLD_PERIOD_MASK == 0 && at_ship_threshold(sd, st, sr) { - yielded = true; - break; + if sd.len() >= next_check { + if crate::columnar::at_serialized_capacity(&( + sd.borrow(), + st.borrow(), + sr.borrow(), + )) { + yielded = true; + break; + } + next_check = sd.len() + check_records; } } yielded @@ -742,7 +723,6 @@ where } fn allocation(chunk: &Self::Chunk) -> (usize, usize, usize) { - use timely::dataflow::channels::ContainerBytes; // Serialized footprint stands in for both `size` and `capacity`: the // chunk owns one logical allocation worth of leaf storage, and we // ship/recycle the whole thing rather than tracking per-leaf diff --git a/src/timely-util/src/columnar/chunk.rs b/src/timely-util/src/columnar/chunk.rs index 28cbd51a11a5f..e2078f2421385 100644 --- a/src/timely-util/src/columnar/chunk.rs +++ b/src/timely-util/src/columnar/chunk.rs @@ -336,6 +336,24 @@ impl ColumnChunk { ColumnChunk::Resident(Rc::new(column), 0) } + /// Append byte-bounded pieces, preserving order and generational depth. + /// A single update may exceed the bound because it cannot be split. + fn push_bounded(column: Column<(D, T, R)>, depth: u8, out: &mut VecDeque) { + let len = column.borrow().len(); + if len <= 1 || column.length_in_bytes() <= COMMIT_BYTES { + if len > 0 { + out.push_back(Self::Resident(Rc::new(column), depth)); + } + return; + } + let view = column.borrow(); + for range in [0..len / 2, len / 2..len] { + let mut part = <(D, T, R) as Columnar>::Container::default(); + part.extend_from_self(view, range); + Self::push_bounded(Column::Typed(part), depth, out); + } + } + /// The body as an owned column. A spilled body is copied out of the pool /// within this call. A shared resident body is copied. pub fn into_column(self) -> Column<(D, T, R)> { @@ -919,10 +937,25 @@ where } ColumnChunk::Resident(rc, depth) => (rc, depth), }; + if rc.length_in_bytes() > COMMIT_BYTES && rc.borrow().len() > 1 { + let col = Rc::try_unwrap(rc).unwrap_or_else(|rc| copy_column(&rc)); + let mut pieces = VecDeque::new(); + Self::push_bounded(col, depth, &mut pieces); + for piece in pieces.into_iter().rev() { + input.push_front(piece); + } + continue; + } let full = at_commit_size(&rc); // A sub-threshold chunk coalesces into the open carry by borrow, // never unwrapping a shared body. - if !full && let Some((mut acc, acc_depth)) = carry.take() { + let fits = carry.as_ref().is_some_and(|(acc, _)| { + acc.length_in_bytes().saturating_add(rc.length_in_bytes()) <= COMMIT_BYTES + }); + if !full + && fits + && let Some((mut acc, acc_depth)) = carry.take() + { let Column::Typed(acc_c) = &mut acc else { unreachable!("carry is always Typed"); }; @@ -1139,6 +1172,7 @@ where /// raw input columns through a [`ColumnChunker`] and wraps its output chunks. pub struct ChunkChunker { inner: ColumnChunker<(D, T, R)>, + ready: VecDeque>, staged: ColumnChunk, } @@ -1152,6 +1186,7 @@ where fn default() -> Self { Self { inner: Default::default(), + ready: VecDeque::new(), staged: Default::default(), } } @@ -1179,14 +1214,20 @@ where type Container = ColumnChunk; fn extract(&mut self) -> Option<&mut Self::Container> { - let col = self.inner.extract()?; - self.staged = ColumnChunk::from_column(std::mem::take(col)); + if self.ready.is_empty() { + let col = self.inner.extract()?; + ColumnChunk::push_bounded(std::mem::take(col), 0, &mut self.ready); + } + self.staged = self.ready.pop_front()?; Some(&mut self.staged) } fn finish(&mut self) -> Option<&mut Self::Container> { - let col = self.inner.finish()?; - self.staged = ColumnChunk::from_column(std::mem::take(col)); + if self.ready.is_empty() { + let col = self.inner.finish()?; + ColumnChunk::push_bounded(std::mem::take(col), 0, &mut self.ready); + } + self.staged = self.ready.pop_front()?; Some(&mut self.staged) } } @@ -1585,6 +1626,95 @@ mod tests { collected } + type WideUpdate = ((u64, String), u64, i64); + type WideChunk = ColumnChunk<(u64, String), u64, i64>; + + fn wide_column(keys: impl Iterator, bytes: usize) -> Column { + let mut column = Column::default(); + for key in keys { + column.push_into(&((key, "x".repeat(bytes)), 0, 1)); + } + column + } + + fn assert_wide_byte_bound(chunks: VecDeque, expected: usize, payload_bytes: usize) { + let mut keys = Vec::new(); + for chunk in chunks { + let column = chunk.into_column(); + assert!( + column.length_in_bytes() <= COMMIT_BYTES || column.borrow().len() == 1, + "{} bytes in a {}-record chunk", + column.length_in_bytes(), + column.borrow().len(), + ); + let view = column.borrow(); + for index in 0..view.len() { + let ((key, payload), time, diff) = view.get(index); + assert_eq!(payload.len(), payload_bytes); + assert!(payload.iter().all(|byte| *byte == b'x')); + assert_eq!((*time, *diff), (0, 1)); + keys.push(*key); + } + } + assert_eq!(keys, (0..u64::cast_from(expected)).collect::>()); + } + + #[mz_ore::test] + fn chunker_enforces_byte_bound() { + let mut chunker = ChunkChunker::default(); + let mut input = wide_column((0..4000).rev(), 3000); + chunker.push_into(&mut input); + let mut chunks = VecDeque::new(); + if let Some(chunk) = chunker.extract() { + chunks.push_back(std::mem::take(chunk)); + } + while let Some(chunk) = chunker.finish() { + chunks.push_back(std::mem::take(chunk)); + } + assert_wide_byte_bound(chunks, 4000, 3000); + } + + #[mz_ore::test] + fn merge_settle_enforces_byte_bound() { + let mut left = VecDeque::from([WideChunk::from_column(wide_column( + (0..2000).step_by(2), + 2100, + ))]); + let mut right = VecDeque::from([WideChunk::from_column(wide_column( + (1..2000).step_by(2), + 2100, + ))]); + let mut merged = VecDeque::new(); + while !left.is_empty() && !right.is_empty() { + WideChunk::merge(&mut left, &mut right, &mut merged); + } + merged.append(&mut left); + merged.append(&mut right); + let mut settled = VecDeque::new(); + WideChunk::settle(&mut merged, true, &mut settled); + assert_wide_byte_bound(settled, 2000, 2100); + } + + #[mz_ore::test] + fn settle_enforces_byte_bound_after_coalescing() { + let mut input = VecDeque::from([ + WideChunk::from_column(wide_column(0..400, 3000)), + WideChunk::from_column(wide_column(400..800, 3000)), + ]); + let mut settled = VecDeque::new(); + WideChunk::settle(&mut input, true, &mut settled); + assert_wide_byte_bound(settled, 800, 3000); + } + + #[mz_ore::test] + fn settle_byte_bound_allows_indivisible_update() { + let mut input = + VecDeque::from([WideChunk::from_column(wide_column(0..1, 2 * COMMIT_BYTES))]); + let mut settled = VecDeque::new(); + WideChunk::settle(&mut input, true, &mut settled); + assert_wide_byte_bound(settled, 1, 2 * COMMIT_BYTES); + } + /// Advancing a large input cuts the output into several chunks near the /// ship threshold, and their concatenation is the reference result. #[mz_ore::test] @@ -1880,7 +2010,7 @@ mod tests { #[cfg_attr(miri, ignore)] // too slow fn settle_commits_at_accumulated_depth() { set_spill_override(Some(test_pool())); - let big: Vec = (0..100_000u64).map(|i| ((i, 0), 0, 1i64)).collect(); + let big: Vec = (0..60_000u64).map(|i| ((i, 0), 0, 1i64)).collect(); let mut input = VecDeque::from([ ColumnChunk::Resident(Rc::new(build_column(&big)), 1), ColumnChunk::Resident(Rc::new(build_column(&[((0, 0), 0, 1)])), 0), @@ -1903,10 +2033,8 @@ mod tests { #[mz_ore::test] #[cfg_attr(miri, ignore)] // too slow fn settle_carry_commits_at_target() { - // ~1.5 MiB per chunk (a row serializes to 32 bytes): under - // `at_commit_size`, so the carry has to coalesce, and a coalesced - // pair lands in the dead zone of the periodic window check. - let chunk_rows = u64::cast_from(1_500_000usize / 32); + // Two inputs fit in one slot. A third must start another chunk. + let chunk_rows = u64::cast_from(800_000usize / 32); let mut input: VecDeque = (0..4u64) .map(|c| { let data: Vec = (0..chunk_rows) @@ -1923,8 +2051,8 @@ mod tests { for chunk in &out { let col = chunk.clone().into_column(); assert!( - col.length_in_bytes() < 2 * COMMIT_BYTES, - "settled chunk of {} bytes exceeds twice the commit target", + col.length_in_bytes() <= COMMIT_BYTES, + "settled chunk of {} bytes exceeds the commit target", col.length_in_bytes(), ); } From bb1979aec9c5bcabdce29ef9d2ee99660a19dea6 Mon Sep 17 00:00:00 2001 From: Dov Alperin Date: Tue, 8 Sep 2026 15:33:38 -0400 Subject: [PATCH 02/31] pool: bound insertion debt during concurrent enforcement Reserve resident bytes before filling pool slots. If enforcement cannot make room within the bounded allowance, write the insertion directly to an extent and expose that fallback in pool metrics. This prevents an occupied single-flight enforcer from allowing an unbounded queue of resident insertions. Add deterministic tests for a stalled enforcer and concurrent fills, including payload round trips and accounting after drops. --- doc/user/data/metrics.yml | 4 + src/ore/src/pool.rs | 148 +++++++++++++++++++-- src/timely-util/src/pool_config/metrics.rs | 1 + 3 files changed, 144 insertions(+), 9 deletions(-) diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index d42be02e4edd0..8b32bc164e62e 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -556,6 +556,10 @@ metrics: help: Evicted chunks re-admitted by an admitting read stealing the slot of a clean backed victim of the same size class. source: src/timely-util/src/pool_config/metrics.rs visibility: internal +- name: mz_column_pool_direct_extent_inserts_total + help: Inserts written directly to an extent because resident admission was full. + source: src/timely-util/src/pool_config/metrics.rs + visibility: internal - name: mz_column_pool_eager_backs_total help: Chunks eagerly compressed to compressed-but-resident by idle spill threads; their later eviction is a pure page release. source: src/timely-util/src/pool_config/metrics.rs diff --git a/src/ore/src/pool.rs b/src/ore/src/pool.rs index e985f57b52029..0eadc89e241cc 100644 --- a/src/ore/src/pool.rs +++ b/src/ore/src/pool.rs @@ -194,6 +194,8 @@ enum Residency { pub struct PoolStats { /// Chunks inserted. pub inserts: u64, + /// Inserts written directly to an extent because resident admission was full. + pub direct_extent_inserts: u64, /// Chunks freed (handle dropped). pub frees: u64, /// Backing writes elided: chunks dead before their compression @@ -279,6 +281,7 @@ pub struct PoolStats { #[derive(Debug, Default)] struct Counters { + direct_extent_inserts: AtomicU64, inserts: AtomicU64, spill_scheduled: AtomicU64, spill_cancelled: AtomicU64, @@ -570,10 +573,13 @@ impl Pool { } /// Allocates a chunk of `len` words and fills it in place: `fill` - /// receives the chunk's slot memory directly and must overwrite all of - /// it (the slot's prior contents are unspecified), so serialization - /// writes its single copy straight into pool memory. The returned handle - /// starts `UnbackedResident`. A zero `len` returns a length-0 handle + /// receives `len` contiguous words and must overwrite all of them. + /// With resident admission, these are the slot's unspecified prior + /// contents, so serialization writes directly into pool memory. + /// Otherwise, `fill` writes into staging for a synchronous extent write. + /// The returned handle + /// starts `UnbackedResident` when admission has room, otherwise `Evicted` + /// with a directly written extent. A zero `len` returns a length-0 handle /// holding no slot; payloads beyond the largest size class fall back to /// a plain heap allocation, always resident, a prototype limitation. /// `hints` steer eviction and write-behind policy; callers without @@ -620,14 +626,22 @@ impl Pool { .oversize_payloads .fetch_add(1, Ordering::Relaxed); } + if class.is_some() { + if !inner.reserve_insert(len_bytes) { + inner.enforce_budget(); + if !inner.reserve_insert(len_bytes) { + return self.insert_extent(len, hints, codec, fill); + } + } + } else { + inner + .counters + .resident_bytes + .fetch_add(u64::cast_from(len_bytes), Ordering::Relaxed); + } // A class with no free slot degrades to the heap path below: an // unpageable chunk beats a dead replica. let slot = class.and_then(|class| inner.alloc_slot(class, len_bytes)); - // Whichever home the payload found, it is resident. - inner - .counters - .resident_bytes - .fetch_add(u64::cast_from(len_bytes), Ordering::Relaxed); let meta = match (class, slot) { (Some(class), Some(slot)) => { let region = &inner.regions[class]; @@ -689,11 +703,49 @@ impl Pool { ChunkHandle { meta } } + fn insert_extent( + &self, + len: usize, + hints: ChunkHints, + codec: &'static dyn ExtentCodec, + fill: impl FnOnce(&mut [u64]), + ) -> ChunkHandle { + // Compress synchronously to keep denied insertions from queuing + // uncompressed payloads behind an occupied enforcer. + let mut words = vec![0; len]; + fill(&mut words); + let inner = &self.0; + let extent = SwapExtent::write(&inner.extent_arena, &words, codec, Scratch::Shrink); + drop(words); + let meta = Arc::new(ChunkMeta::new( + inner, + len, + region::size_class_for(len * 8), + hints.depth, + codec, + Residency::Evicted, + None, + None, + )); + inner.live_chunks.fetch_add(1, Ordering::Relaxed); + inner + .counters + .direct_extent_inserts + .fetch_add(1, Ordering::Relaxed); + { + let mut state = meta.state(); + inner.commit_extent(&meta, &mut state, extent); + } + inner.enforce_or_defer_compressed_cap(); + ChunkHandle { meta } + } + /// Snapshot of the pool's counters. pub fn stats(&self) -> PoolStats { let c = &self.0.counters; PoolStats { inserts: c.inserts.load(Ordering::Relaxed), + direct_extent_inserts: c.direct_extent_inserts.load(Ordering::Relaxed), frees: c.frees.load(Ordering::Relaxed), writes_elided: c.writes_elided.load(Ordering::Relaxed), evictions_compress: c.evictions_compress.load(Ordering::Relaxed), @@ -986,6 +1038,23 @@ impl PoolInner { } } + /// Reserve insertion bytes before populating a slot. Enforcement can + /// lag by one eighth of the budget, or one payload for small budgets. + /// Read admissions use the budget itself and cannot consume this slack. + fn reserve_insert(&self, len_bytes: usize) -> bool { + let len = u64::cast_from(len_bytes); + self.counters + .resident_bytes + .try_update(Ordering::Relaxed, Ordering::Relaxed, |cur| { + let budget = self.budget_bytes.load(Ordering::Relaxed); + let ceiling = budget.saturating_add((budget / 8).max(len)); + let next = cur.checked_add(len)?; + let oversize = self.counters.oversize_bytes.load(Ordering::Relaxed); + (next.saturating_sub(oversize) <= ceiling).then_some(next) + }) + .is_ok() + } + fn enforce_budget(&self) { // Single-flight: enforcement runs synchronously on whichever thread // trips it (every insert), and concurrent passes would @@ -3172,6 +3241,67 @@ mod tests { ); } + #[mz_ore::test] + fn insertion_debt_is_bounded_during_enforcement() { + let budget = 2 * SMALL * 8; + let pool = test_pool(budget); + let guard = pool.0.enforcing.lock().expect("enforcement lock"); + let mut handles = Vec::new(); + for seed in 0..16 { + handles.push(insert(&pool, &mut payload(SMALL, seed))); + assert!( + pool.stats().resident_bytes <= u64::cast_from(budget + SMALL * 8), + "an occupied enforcer must not allow unlimited insertion debt", + ); + } + drop(guard); + for (seed, handle) in handles.iter().enumerate() { + assert_eq!(read(handle), payload(SMALL, u64::cast_from(seed))); + } + drop(handles); + assert_eq!(pool.stats().resident_bytes, 0); + assert_eq!(pool.stats().live_chunks, 0); + assert_eq!(pool.stats().extent_resident_bytes, 0); + } + + #[mz_ore::test] + fn admission_reserves_before_concurrent_fills() { + let budget = 2 * SMALL * 8; + let pool = test_pool(budget); + let guard = pool.0.enforcing.lock().expect("enforcement lock"); + let gate = Arc::new(std::sync::Barrier::new(9)); + let threads: Vec<_> = (0..8u64) + .map(|seed| { + let pool = pool.clone(); + let gate = Arc::clone(&gate); + std::thread::spawn(move || { + pool.insert_with(SMALL, ChunkHints::default(), &TEST_CODEC, |dst| { + gate.wait(); + gate.wait(); + dst.copy_from_slice(&payload(SMALL, seed)); + }) + }) + }) + .collect(); + gate.wait(); + let reserved = pool.stats().resident_bytes; + // Release every producer even if the assertion fails. + gate.wait(); + let handles: Vec<_> = threads + .into_iter() + .map(|t| t.join().expect("producer panicked")) + .collect(); + drop(guard); + assert!(reserved <= u64::cast_from(budget + SMALL * 8)); + assert!(pool.stats().direct_extent_inserts > 0); + for (seed, handle) in handles.iter().enumerate() { + assert_eq!(read(handle), payload(SMALL, u64::cast_from(seed))); + } + drop(handles); + assert_eq!(pool.stats().resident_bytes, 0); + assert_eq!(pool.stats().live_chunks, 0); + } + #[mz_ore::test] fn set_budget_retunes_in_place() { let pool = test_pool(usize::MAX); diff --git a/src/timely-util/src/pool_config/metrics.rs b/src/timely-util/src/pool_config/metrics.rs index 1146a721d8252..7221f38a79b03 100644 --- a/src/timely-util/src/pool_config/metrics.rs +++ b/src/timely-util/src/pool_config/metrics.rs @@ -39,6 +39,7 @@ pub fn register(registry: &MetricsRegistry) { // instantaneous levels are mixed under the one gauge type, so the // `_total` name suffix, not the metric type, marks a field as // monotonic. + gauge(registry, metric!(name: "mz_column_pool_direct_extent_inserts_total", help: "Inserts written directly to an extent because resident admission was full."), |s| s.direct_extent_inserts); gauge(registry, metric!(name: "mz_column_pool_resident_bytes", help: "Uncompressed bytes resident in the buffer pool."), |s| s.resident_bytes); gauge(registry, metric!(name: "mz_column_pool_oversize_bytes", help: "Bytes held by oversize chunks that bypass pool paging."), |s| s.oversize_bytes); gauge(registry, metric!(name: "mz_column_pool_inserts_total", help: "Chunks inserted into the buffer pool."), |s| s.inserts); From 761ac84fc4855c61943c0c4ec826fbdb3f19e445 Mon Sep 17 00:00:00 2001 From: Dov Alperin Date: Tue, 8 Sep 2026 15:51:56 -0400 Subject: [PATCH 03/31] storage: offload upsert drain reads to the blocking executor Add enable_upsert_async_reads for chunked stash drains and feedback probes. Read pooled bodies on Tokio's blocking executor with at most eight queued or running reads per pool. Submitted jobs retain their handles and permits through cancellation. Resident columns stay on the timely worker, and merge/seal reads remain synchronous. Default the flag off in production and on in mzcompose defaults, register it for stress-test flag flips, and expose submitted/in-flight read metrics. This is an offload experiment: it adds scheduling and read-buffer allocation overhead, and does not pipeline reads within one drain. Validate cancellation and admission bounds, copy-out without residency admission, straddled probe keys, and skipped chunks. Compare paged, synchronous chunked, and asynchronous chunked output in the upsert operator scenarios, including forced pool eviction. Validation: 302 pool/timely utility tests passed; 14 focused async and upsert operator tests passed after the final harness changes. bin/fmt, flag-registration lint, storage cargo check, and clippy for all targets in mz-ore, mz-timely-util, and mz-storage passed. Full bin/lint remains blocked by macOS Bash 3 in check-protobuf and environment stripping in check-mzcompose-files under the jj workspace. No performance results yet. --- doc/user/data/metrics.yml | 8 + misc/python/materialize/mzcompose/__init__.py | 5 + .../materialize/parallel_workload/action.py | 1 + src/ore/src/pool.rs | 170 ++++++++++++++++++ src/storage-types/src/dyncfgs.rs | 11 ++ .../src/upsert_continual_feedback_v2.rs | 62 +++++-- src/timely-util/src/columnar/chunk.rs | 72 +++++++- src/timely-util/src/columnar/unload.rs | 105 +++++++---- src/timely-util/src/pool_config/metrics.rs | 2 + 9 files changed, 380 insertions(+), 56 deletions(-) diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index 8b32bc164e62e..04d8a2228dff1 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -556,6 +556,14 @@ metrics: help: Evicted chunks re-admitted by an admitting read stealing the slot of a clean backed victim of the same size class. source: src/timely-util/src/pool_config/metrics.rs visibility: internal +- name: mz_column_pool_async_reads_in_flight + help: Submitted pool reads that have not released their concurrency permit. + source: src/timely-util/src/pool_config/metrics.rs + visibility: internal +- name: mz_column_pool_async_reads_total + help: Pool reads submitted to the blocking executor. + source: src/timely-util/src/pool_config/metrics.rs + visibility: internal - name: mz_column_pool_direct_extent_inserts_total help: Inserts written directly to an extent because resident admission was full. source: src/timely-util/src/pool_config/metrics.rs diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index a2e156a9e627b..6775050092307 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -364,6 +364,11 @@ def get_variable_system_parameters( "true", ["true", "false"], ), + VariableSystemParameter( + "enable_upsert_async_reads", + "true", + ["true", "false"], + ), VariableSystemParameter( "enable_upsert_v2", "false", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 6a008fc235834..5e77cdc17e42f 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -3179,6 +3179,7 @@ def __init__( ] self.flags_with_values["enable_upsert_paged_spill"] = BOOLEAN_FLAG_VALUES self.flags_with_values["enable_upsert_chunked_stash"] = BOOLEAN_FLAG_VALUES + self.flags_with_values["enable_upsert_async_reads"] = BOOLEAN_FLAG_VALUES self.flags_with_values["column_chunk_compress_min_depth"] = [ "0", # compress every spilled body "1", # the default: fresh chunks store uncompressed diff --git a/src/ore/src/pool.rs b/src/ore/src/pool.rs index 0eadc89e241cc..b8ac6d167af41 100644 --- a/src/ore/src/pool.rs +++ b/src/ore/src/pool.rs @@ -80,6 +80,11 @@ use crate::pool::region::{Region, SIZE_CLASSES}; /// NOTE: Seen OoMs with Miri since it actually allocates the capacity. const CLASS_CAPACITY_BYTES: usize = if cfg!(miri) { 16 << 20 } else { 1 << 40 }; +// At most eight queued or running reads per pool. Completed buffers belong +// to callers, whose staging limits must bound their retained memory. +#[cfg(feature = "async")] +const ASYNC_READ_CONCURRENCY: usize = 8; + /// A chunk-provided transform between a chunk's body bytes and the stored /// bytes its extent holds. The pool owns scheduling: spill threads, the /// residency state machine, cancellation, and the ledger. It invokes the @@ -196,6 +201,10 @@ pub struct PoolStats { pub inserts: u64, /// Inserts written directly to an extent because resident admission was full. pub direct_extent_inserts: u64, + /// Reads submitted to the blocking executor. + pub async_reads: u64, + /// Submitted reads that have not released their concurrency permit. + pub async_reads_in_flight: u64, /// Chunks freed (handle dropped). pub frees: u64, /// Backing writes elided: chunks dead before their compression @@ -282,6 +291,7 @@ pub struct PoolStats { #[derive(Debug, Default)] struct Counters { direct_extent_inserts: AtomicU64, + async_reads: AtomicU64, inserts: AtomicU64, spill_scheduled: AtomicU64, spill_cancelled: AtomicU64, @@ -381,6 +391,8 @@ struct PoolInner { /// counter read still has its bytes enforced rather than dropped. enforce_pending: std::sync::atomic::AtomicBool, counters: Counters, + #[cfg(feature = "async")] + read_slots: Arc, spill: Spill, } @@ -568,6 +580,8 @@ impl Pool { enforcing: Mutex::new(()), enforce_pending: std::sync::atomic::AtomicBool::new(false), counters: Counters::default(), + #[cfg(feature = "async")] + read_slots: Arc::new(tokio::sync::Semaphore::new(ASYNC_READ_CONCURRENCY)), spill: Spill::default(), }))) } @@ -746,6 +760,17 @@ impl Pool { PoolStats { inserts: c.inserts.load(Ordering::Relaxed), direct_extent_inserts: c.direct_extent_inserts.load(Ordering::Relaxed), + async_reads: c.async_reads.load(Ordering::Relaxed), + async_reads_in_flight: { + #[cfg(feature = "async")] + { + u64::cast_from(ASYNC_READ_CONCURRENCY - self.0.read_slots.available_permits()) + } + #[cfg(not(feature = "async"))] + { + 0 + } + }, frees: c.frees.load(Ordering::Relaxed), writes_elided: c.writes_elided.load(Ordering::Relaxed), evictions_compress: c.evictions_compress.load(Ordering::Relaxed), @@ -1958,6 +1983,37 @@ impl ChunkHandle { self.read_impl(0..self.meta.len, dst, false); } + /// Copy this chunk on the blocking executor without admitting it to the pool. + /// + /// Requires a Tokio runtime. A submitted read retains its handle and + /// concurrency permit until it finishes, even if the caller cancels. + /// The returned buffer belongs to the caller. + #[cfg(feature = "async")] + pub async fn read_async(self: &Arc) -> Vec { + let permit = Arc::clone(&self.meta.pool.read_slots) + .acquire_owned() + .await + .expect("pool read semaphore remains open"); + let handle = Arc::clone(self); + self.meta + .pool + .counters + .async_reads + .fetch_add(1, Ordering::Relaxed); + crate::task::spawn_blocking( + || "pool_read", + move || { + // A cancelled JoinHandle must not release admission while its + // blocking read still owns a slot or extent reference. + let _permit = permit; + let mut words = Vec::new(); + handle.read_into(&mut words); + words + }, + ) + .await + } + /// As [`ChunkHandle::read_into`], restricted to the word range `range` /// of the chunk's contents, which must lie within them. `dst` receives /// exactly the range. @@ -2216,6 +2272,9 @@ impl Drop for ChunkHandle { #[cfg(test)] mod tests { + #[cfg(feature = "async")] + use std::sync::atomic::{AtomicBool, AtomicUsize}; + use super::*; use crate::pool::extent::TEST_CODEC; @@ -2231,6 +2290,117 @@ mod tests { pool } + #[cfg(feature = "async")] + #[derive(Debug, Default)] + struct DelayedReadCodec { + entered: AtomicUsize, + released: AtomicBool, + } + + #[cfg(feature = "async")] + impl ExtentCodec for DelayedReadCodec { + fn encode(&self, body: &[u8], out: &mut Vec) { + TEST_CODEC.encode(body, out); + } + + fn decode(&self, stored: &[u8], body: &mut [u8]) { + // NOTE: Test-only delay models a page fault while the state lock is held. + self.entered.fetch_add(1, Ordering::SeqCst); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + while !self.released.load(Ordering::SeqCst) { + assert!( + std::time::Instant::now() < deadline, + "test read was released" + ); + std::thread::sleep(std::time::Duration::from_millis(1)); + } + TEST_CODEC.decode(stored, body); + } + } + + #[cfg(feature = "async")] + struct ReleaseReadsOnDrop(&'static DelayedReadCodec); + + #[cfg(feature = "async")] + impl Drop for ReleaseReadsOnDrop { + fn drop(&mut self) { + self.0.released.store(true, Ordering::SeqCst); + } + } + + #[cfg(feature = "async")] + async fn wait_for_read_state(mut ready: impl FnMut() -> bool) { + tokio::time::timeout(std::time::Duration::from_secs(10), async { + while !ready() { + tokio::task::yield_now().await; + } + }) + .await + .expect("read workers made progress"); + } + + #[cfg(feature = "async")] + #[mz_ore::test(tokio::test)] + #[cfg_attr(miri, ignore)] + async fn async_read_round_trip_without_admission() { + for budget in [0, usize::MAX] { + let pool = test_pool(budget); + let expected = payload(8192, 42); + let handle = Arc::new(insert(&pool, &mut expected.clone())); + let resident = pool.stats().resident_bytes; + assert_eq!(handle.read_async().await, expected); + assert_eq!(pool.stats().resident_bytes, resident); + assert_eq!(pool.stats().async_reads, 1); + assert_eq!(pool.stats().async_reads_in_flight, 0); + } + } + + #[cfg(feature = "async")] + #[mz_ore::test(tokio::test)] + #[cfg_attr(miri, ignore)] + async fn async_reads_remain_bounded_after_cancellation() { + let pool = test_pool(0); + let codec = Box::leak(Box::new(DelayedReadCodec::default())); + let release = ReleaseReadsOnDrop(codec); + let expected = payload(8192, 7); + let handles: Vec<_> = (0..=ASYNC_READ_CONCURRENCY) + .map(|_| { + Arc::new( + pool.insert_with(expected.len(), ChunkHints::default(), codec, |dst| { + dst.copy_from_slice(&expected); + }), + ) + }) + .collect(); + let mut reads: Vec<_> = handles.iter().map(|h| Box::pin(h.read_async())).collect(); + for read in &mut reads { + assert!(futures::poll!(read.as_mut()).is_pending()); + } + wait_for_read_state(|| codec.entered.load(Ordering::SeqCst) == ASYNC_READ_CONCURRENCY) + .await; + assert_eq!( + pool.stats().async_reads, + u64::cast_from(ASYNC_READ_CONCURRENCY) + ); + assert_eq!( + pool.stats().async_reads_in_flight, + u64::cast_from(ASYNC_READ_CONCURRENCY) + ); + drop(reads); + drop(handles); + // The unsubmitted ninth read frees immediately. Detached blocking + // jobs must retain both their handles and their admission permits. + assert_eq!(pool.stats().frees, 1); + assert_eq!( + pool.stats().async_reads_in_flight, + u64::cast_from(ASYNC_READ_CONCURRENCY) + ); + drop(release); + wait_for_read_state(|| pool.stats().frees == u64::cast_from(ASYNC_READ_CONCURRENCY + 1)) + .await; + assert_eq!(pool.stats().async_reads_in_flight, 0); + } + /// Scales an iteration count down under Miri, where one interpreted /// compression costs what thousands do natively. fn rounds(native: u64, miri: u64) -> u64 { diff --git a/src/storage-types/src/dyncfgs.rs b/src/storage-types/src/dyncfgs.rs index a98278efa5621..e27dff344b92e 100644 --- a/src/storage-types/src/dyncfgs.rs +++ b/src/storage-types/src/dyncfgs.rs @@ -442,6 +442,16 @@ pub const ENABLE_UPSERT_CHUNKED_STASH: Config = Config::new( ParameterScope::Replica, ); +/// Offload chunked upsert stash drains and feedback lookups to the blocking +/// executor. Merge reads remain synchronous. Read at operator construction. +pub const ENABLE_UPSERT_ASYNC_READS: Config = Config::new( + "enable_upsert_async_reads", + false, + "Read sealed stash and feedback chunks asynchronously when enable_upsert_v2 and \ + enable_upsert_chunked_stash are true. Takes effect on new dataflows.", + ParameterScope::Replica, +); + // RocksDB /// How many times to try to cleanup old RocksDB DB's on disk before giving up. @@ -566,6 +576,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&SUSPENDABLE_SOURCES) .add(&ENABLE_UPSERT_PAGED_SPILL) .add(&ENABLE_UPSERT_CHUNKED_STASH) + .add(&ENABLE_UPSERT_ASYNC_READS) .add(&WALLCLOCK_GLOBAL_LAG_HISTOGRAM_RETENTION_INTERVAL) .add(&WALLCLOCK_LAG_HISTORY_RETENTION_INTERVAL) .add(&crate::sources::sql_server::CDC_CLEANUP_CHANGE_TABLE) diff --git a/src/storage/src/upsert_continual_feedback_v2.rs b/src/storage/src/upsert_continual_feedback_v2.rs index 5b4b699ac2048..d00b818cc9379 100644 --- a/src/storage/src/upsert_continual_feedback_v2.rs +++ b/src/storage/src/upsert_continual_feedback_v2.rs @@ -105,7 +105,7 @@ use mz_repr::{Datum, Diff, GlobalId, Row}; #[cfg(feature = "fuzzing")] use mz_row_spine::DatumSeq; use mz_row_spine::{ValRowColPagedBuilder, ValRowSpine}; -use mz_storage_types::dyncfgs::ENABLE_UPSERT_CHUNKED_STASH; +use mz_storage_types::dyncfgs::{ENABLE_UPSERT_ASYNC_READS, ENABLE_UPSERT_CHUNKED_STASH}; use mz_storage_types::errors::{DataflowError, EnvelopeError, UpsertError}; use mz_timely_util::builder_async::{ AsyncOutputHandle, Event as AsyncEvent, OperatorBuilder as AsyncOperatorBuilder, @@ -148,7 +148,7 @@ pub enum UpsertStashFlavor { Paged, /// Chunk merge batcher stash, chunk-spine feedback arrangement, /// bulk-probe drain. Spills through the process buffer pool. - Chunked, + Chunked { async_reads: bool }, } impl UpsertStashFlavor { @@ -157,7 +157,9 @@ impl UpsertStashFlavor { /// for its whole life even if the flag flips underneath it. pub fn from_config(config: &ConfigSet) -> Self { if ENABLE_UPSERT_CHUNKED_STASH.get(config) { - Self::Chunked + Self::Chunked { + async_reads: ENABLE_UPSERT_ASYNC_READS.get(config), + } } else { Self::Paged } @@ -442,7 +444,7 @@ where source_config.source_statistics.clone(), ); match flavor { - UpsertStashFlavor::Chunked => { + UpsertStashFlavor::Chunked { async_reads } => { // Chains and sealed batches alike are `FeedbackChunk`s whose // bodies spill to the buffer pool, behind the same process spill // gate as the source stash. @@ -461,6 +463,7 @@ where persist_token, upsert_metrics, source_config, + async_reads, ) } UpsertStashFlavor::Paged => { @@ -484,6 +487,7 @@ where persist_token, upsert_metrics, source_config, + false, ) } } @@ -568,6 +572,7 @@ fn build_upsert_operator<'scope, A, T, FromTime>( persist_token: Option>, upsert_metrics: UpsertMetrics, source_config: crate::source::SourceExportCreationConfig, + async_reads: bool, ) -> ( VecCollection<'scope, T, Result, Diff>, StreamVec<'scope, T, (Option, HealthStatusUpdate)>, @@ -806,6 +811,7 @@ where &mut persist_trace, source_config.worker_id, source_config.id, + async_reads, ) .await; @@ -926,6 +932,7 @@ where trace: &mut TraceAgent, worker_id: usize, source_id: GlobalId, + async_reads: bool, ) -> DrainStats; } @@ -970,9 +977,10 @@ where trace: &mut TraceAgent, worker_id: usize, source_id: GlobalId, + async_reads: bool, ) -> DrainStats { drain_sealed_input_chunked( - sealed.into_iter().map(ColumnChunk::into_column), + sealed.into_iter(), ineligible, output_handle, output_cap, @@ -980,6 +988,7 @@ where trace, worker_id, source_id, + async_reads, ) .await } @@ -1020,6 +1029,7 @@ where trace: &mut TraceAgent, worker_id: usize, source_id: GlobalId, + _async_reads: bool, ) -> DrainStats { drain_sealed_input_paged( sealed, @@ -1104,7 +1114,7 @@ struct DrainStats { /// probe hits for its keys) is resident regardless of drain size. Only the /// re-stashed ineligible set is materialized. async fn drain_sealed_input_chunked( - sealed: impl Iterator>>, + sealed: impl Iterator>, ineligible: &mut Vec>, output_handle: &UpsertOutputHandle, output_cap: &Capability, @@ -1112,6 +1122,7 @@ async fn drain_sealed_input_chunked( trace: &mut TraceAgent>, worker_id: usize, source_id: GlobalId, + async_reads: bool, ) -> DrainStats where T: Timestamp + TotalOrder + Lattice + Sync, @@ -1141,6 +1152,11 @@ where for chunk in sealed { use columnar::{Index, Len}; + let chunk = if async_reads { + chunk.into_column_async().await + } else { + chunk.into_column() + }; let view = chunk.borrow(); let total = view.len(); let mut start = 0; @@ -1184,7 +1200,13 @@ where use columnar::Borrow; let mut staging = as columnar::Columnar>::Container::default(); for batch in &batches { - batch.extract_into(probe_col.borrow(), &mut staging); + if async_reads { + batch + .extract_into_async(probe_col.borrow(), &mut staging) + .await; + } else { + batch.extract_into(probe_col.borrow(), &mut staging); + } } let staged = staging.borrow(); let mut hits: Vec<_> = (0..staged.len()) @@ -1499,10 +1521,8 @@ mod test { Row::pack_slice(&[Datum::Int64(k), Datum::Int64(v)]) } - // Runs the test body once per stash flavor and asserts the two flavors - // produce identical (consolidated) output, so every scenario covers both - // operator arms. Returns one flavor's output for the caller's own - // expected-value assertion. + // Compare paged, synchronous chunked, and asynchronous chunked output + // before checking each scenario's expected result. macro_rules! upsert_test { (|$input:ident, $persist:ident, $worker:ident| $body:block) => {{ let run = |flavor: UpsertStashFlavor| { @@ -1571,8 +1591,10 @@ mod test { }; let paged = run(UpsertStashFlavor::Paged); - let chunked = run(UpsertStashFlavor::Chunked); + let chunked = run(UpsertStashFlavor::Chunked { async_reads: false }); assert_eq!(paged, chunked, "stash flavors must produce equal output"); + let asynchronous = run(UpsertStashFlavor::Chunked { async_reads: true }); + assert_eq!(chunked, asynchronous, "async reads must preserve output"); chunked }}; } @@ -1751,13 +1773,14 @@ mod test { /// paged flavor (which the harness also runs) routes through the column /// pager rather than the chunk override, so it stays resident and serves /// as the reference. - #[mz_ore::test] + #[mz_ore::test(tokio::test)] #[cfg_attr(miri, ignore)] - fn drain_reads_spilled_chunks() { + async fn drain_reads_spilled_chunks() { use mz_ore::pool::Pool; use mz_timely_util::columnar::chunk::set_spill_override; let pool = Pool::new().expect("pool creation"); + pool.set_budget(0); set_spill_override(Some(pool.clone())); const KEYS: i64 = 1500; @@ -1783,6 +1806,11 @@ mod test { "chunks should have spilled through the pool" ); + assert!( + pool.stats().async_reads > 0, + "the async drain must read spilled chunks" + ); + let mut expected: Vec<(Result, _, _)> = Vec::new(); for k in 0..KEYS { expected.push((Ok(row(k, k)), new_ts(1), Diff::MINUS_ONE)); @@ -1961,7 +1989,11 @@ mod test { #[mz_ore::test] #[cfg_attr(miri, ignore)] fn lagging_replacement_below_upper_strands_data() { - for flavor in [UpsertStashFlavor::Paged, UpsertStashFlavor::Chunked] { + for flavor in [ + UpsertStashFlavor::Paged, + UpsertStashFlavor::Chunked { async_reads: false }, + UpsertStashFlavor::Chunked { async_reads: true }, + ] { let (frontier, emitted) = run_below_upper_scenario_v2(flavor); // The below-upper data is discarded (no output) and the output diff --git a/src/timely-util/src/columnar/chunk.rs b/src/timely-util/src/columnar/chunk.rs index e2078f2421385..4c39d35712f66 100644 --- a/src/timely-util/src/columnar/chunk.rs +++ b/src/timely-util/src/columnar/chunk.rs @@ -44,6 +44,7 @@ use std::cell::Cell; use std::cell::RefCell; use std::collections::VecDeque; use std::rc::Rc; +use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use columnar::bytes::indexed; @@ -281,7 +282,7 @@ pub struct SpilledBody { /// and miss every path that skips it. compressed: bool, /// The pool chunk holding the serialized column. - handle: ChunkHandle, + handle: Arc, } /// A sorted, consolidated run of `(D, T, R)` updates, resident or spilled. @@ -369,6 +370,14 @@ impl ColumnChunk { } } + /// Load a spilled body asynchronously, keeping resident columns on this worker. + pub async fn into_column_async(self) -> Column<(D, T, R)> { + match self { + Self::Spilled(body, _) => Column::Align(body.handle.read_async().await), + resident @ Self::Resident(..) => resident.into_column(), + } + } + /// True when the body lives in the pool. pub fn is_spilled(&self) -> bool { matches!(self, ColumnChunk::Spilled(_, _)) @@ -446,7 +455,7 @@ impl ColumnChunk { time_lower, time_upper: time_upper.into(), compressed, - handle, + handle: Arc::new(handle), }), depth, ) @@ -1094,6 +1103,22 @@ where } } + async fn extract_into_async( + &self, + probes: Self::Probes<'_>, + probe_index: &mut usize, + staging: &mut Self::Staging, + ) { + match self { + Self::Resident(_, _) => self.extract_into(probes, probe_index, staging), + Self::Spilled(body, _) => { + let words = body.handle.read_async().await; + let view = borrow_words::<((K, V), T, R)>(&words); + extract_view_into::(view, probes, probe_index, staging); + } + } + } + fn fetch_into(&self, staging: &mut Self::Staging) { match self { ColumnChunk::Resident(col, _) => { @@ -1592,6 +1617,49 @@ mod tests { } } + #[mz_ore::test(tokio::test)] + #[cfg_attr(miri, ignore)] + async fn async_reads_preserve_straddles_and_skip_unprobed_chunks() { + let input = vec![ + ((1, 0), 0, 1), + ((2, 0), 0, 1), + ((2, 1), 0, -1), + ((3, 0), 0, 1), + ((8, 0), 0, 1), + ]; + for spill in [false, true] { + let pool = Pool::new().expect("pool creation"); + pool.set_budget(0); + let chunks: Vec = if spill { + chunked_spilled(&input, &[1, 1], &pool).into() + } else { + chunked(&input, &[1, 1]).into() + }; + let description = Description::new( + Antichain::from_elem(0u64), + Antichain::new(), + Antichain::from_elem(0u64), + ); + let batch = ChunkBatch::new(chunks.clone(), description); + let mut probes = ::Container::default(); + for key in [0u64, 2, 4, 9] { + probes.push(key); + } + let mut staging = ::Container::default(); + batch + .extract_into_async(probes.borrow(), &mut staging) + .await; + assert_eq!(collect_staging(&staging), input[1..3]); + assert_eq!(pool.stats().async_reads, if spill { 2 } else { 0 }); + let mut actual = Vec::new(); + for chunk in chunks { + actual.append(&mut collect_column(&chunk.into_column_async().await)); + } + assert_eq!(actual, input); + assert_eq!(pool.stats().resident_bytes, 0); + } + } + /// `locate` answers the three-way span comparison for every probe /// placement: below, within, and past the chunk's keys. #[mz_ore::test] diff --git a/src/timely-util/src/columnar/unload.rs b/src/timely-util/src/columnar/unload.rs index 58911a62bb5e9..e257eb4b8861c 100644 --- a/src/timely-util/src/columnar/unload.rs +++ b/src/timely-util/src/columnar/unload.rs @@ -92,6 +92,16 @@ pub trait UnloadChunk: Chunk { staging: &mut Self::Staging, ); + /// As `extract_into`, allowing a spilled body to be read asynchronously. + fn extract_into_async( + &self, + probes: Self::Probes<'_>, + probe_index: &mut usize, + staging: &mut Self::Staging, + ) -> impl std::future::Future { + async move { self.extract_into(probes, probe_index, staging) } + } + /// Append the whole chunk into `staging` (the scan path). fn fetch_into(&self, staging: &mut Self::Staging); } @@ -110,51 +120,32 @@ pub trait UnloadBatch { /// whose continuation follows in staging. fn extract_into(&self, probes: C::Probes<'_>, staging: &mut C::Staging); + /// Extract probe hits in order, awaiting each selected chunk's read. + fn extract_into_async( + &self, + probes: C::Probes<'_>, + staging: &mut C::Staging, + ) -> impl std::future::Future; + /// Materialize the batch's full contents into `staging` (the scan path). fn fetch_into(&self, staging: &mut C::Staging); } impl UnloadBatch for ChunkBatch { fn extract_into(&self, probes: C::Probes<'_>, staging: &mut C::Staging) { - 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 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; - } - // Consume probes in the gap below this chunk's first key: they - // match nothing in the batch, and deciding so from resident - // metadata is what keeps an untouched body unopened. - while probe_index < count && chunks[chunk].locate(probes, probe_index) == Ordering::Less - { - probe_index += 1; - } - if probe_index < count && chunks[chunk].locate(probes, probe_index) == Ordering::Equal { - 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 re-offer). - chunk += 1; + let (mut probe_index, mut chunk) = (0, 0); + while let Some(next) = next_probe_chunk(&self.chunks, probes, &mut probe_index, &mut chunk) + { + next.extract_into(probes, &mut probe_index, staging); + } + } + + async fn extract_into_async(&self, probes: C::Probes<'_>, staging: &mut C::Staging) { + let (mut probe_index, mut chunk) = (0, 0); + while let Some(next) = next_probe_chunk(&self.chunks, probes, &mut probe_index, &mut chunk) + { + next.extract_into_async(probes, &mut probe_index, staging) + .await; } } @@ -165,6 +156,42 @@ impl UnloadBatch for ChunkBatch { } } +// Select from resident fences only. The caller advances the probe index +// through the returned chunk, leaving its last-key probe for the next one. +fn next_probe_chunk<'a, C: UnloadChunk>( + chunks: &'a [C], + probes: C::Probes<'_>, + probe_index: &mut usize, + chunk: &mut usize, +) -> Option<&'a C> { + let count = C::probe_count(probes); + while *probe_index < count && *chunk < chunks.len() { + let below = |c: usize| chunks[c].locate(probes, *probe_index) == Ordering::Greater; + 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; + } + let next = chunks.get(*chunk)?; + *chunk += 1; + while *probe_index < count && next.locate(probes, *probe_index) == Ordering::Less { + *probe_index += 1; + } + if *probe_index < count && next.locate(probes, *probe_index) == Ordering::Equal { + return Some(next); + } + } + None +} + #[cfg(test)] mod tests { //! Contract tests for the batch driver over a miniature row family: diff --git a/src/timely-util/src/pool_config/metrics.rs b/src/timely-util/src/pool_config/metrics.rs index 7221f38a79b03..9d7d32b4f5049 100644 --- a/src/timely-util/src/pool_config/metrics.rs +++ b/src/timely-util/src/pool_config/metrics.rs @@ -39,6 +39,8 @@ pub fn register(registry: &MetricsRegistry) { // instantaneous levels are mixed under the one gauge type, so the // `_total` name suffix, not the metric type, marks a field as // monotonic. + gauge(registry, metric!(name: "mz_column_pool_async_reads_total", help: "Pool reads submitted to the blocking executor."), |s| s.async_reads); + gauge(registry, metric!(name: "mz_column_pool_async_reads_in_flight", help: "Submitted pool reads that have not released their concurrency permit."), |s| s.async_reads_in_flight); gauge(registry, metric!(name: "mz_column_pool_direct_extent_inserts_total", help: "Inserts written directly to an extent because resident admission was full."), |s| s.direct_extent_inserts); gauge(registry, metric!(name: "mz_column_pool_resident_bytes", help: "Uncompressed bytes resident in the buffer pool."), |s| s.resident_bytes); gauge(registry, metric!(name: "mz_column_pool_oversize_bytes", help: "Bytes held by oversize chunks that bypass pool paging."), |s| s.oversize_bytes); From d8989ee40c926a86e28ffc0c273cc89a1b88c2c3 Mon Sep 17 00:00:00 2001 From: Dov Alperin Date: Tue, 8 Sep 2026 16:05:46 -0400 Subject: [PATCH 04/31] timely-util: compress chunk output directly into extents Add enable_column_chunk_direct_compressed_output, off in production and on in mzcompose test defaults. Eligible spilled chunks encode directly into an actual-size extent without entering the resident slot pool. Preserve the compression depth floor and existing empty/oversize fallbacks. Typed columns use bounded retained serialization scratch; aligned columns encode from their existing words. Expose Pool::insert_cold and a separate cold-insertion metric. Share extent registration and reclamation accounting with admission-denied insertions. Encoding remains synchronous and bypassing resident slots gives up write elision for short-lived chunks. Test cold insertion, read admission, extent reuse, cleanup, empty/oversize handling, byte-identical typed/aligned serialization, and the compression floor. Exercise the upsert drain with direct output disabled/enabled and synchronous/asynchronous reads against the paged reference. Validation: 331 selected utility and upsert tests passed. Clippy passed for all targets in mz-ore, mz-timely-util, mz-storage, and mz-compute. Formatting and flag-registration checks passed. Full lint has local tooling failures in check-protobuf (macOS Bash 3) and check-mzcompose-files (Git environment stripped under jj). Performance remains to be measured. --- doc/user/data/metrics.yml | 4 + misc/python/materialize/mzcompose/__init__.py | 5 + .../materialize/parallel_workload/action.py | 3 + src/compute-types/src/dyncfgs.rs | 10 ++ src/compute/src/compute_state.rs | 3 + src/ore/src/pool.rs | 99 +++++++++++++++- .../src/upsert_continual_feedback_v2.rs | 17 ++- src/timely-util/src/columnar/chunk.rs | 108 +++++++++++++++--- src/timely-util/src/pool_config/metrics.rs | 1 + 9 files changed, 229 insertions(+), 21 deletions(-) diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index 04d8a2228dff1..c105e251d2e38 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -564,6 +564,10 @@ metrics: help: Pool reads submitted to the blocking executor. source: src/timely-util/src/pool_config/metrics.rs visibility: internal +- name: mz_column_pool_cold_inserts_total + help: Chunks inserted directly into an extent by caller request. + source: src/timely-util/src/pool_config/metrics.rs + visibility: internal - name: mz_column_pool_direct_extent_inserts_total help: Inserts written directly to an extent because resident admission was full. source: src/timely-util/src/pool_config/metrics.rs diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 6775050092307..f57c5ea6d1789 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -364,6 +364,11 @@ def get_variable_system_parameters( "true", ["true", "false"], ), + VariableSystemParameter( + "enable_column_chunk_direct_compressed_output", + "true", + ["true", "false"], + ), VariableSystemParameter( "enable_upsert_async_reads", "true", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 5e77cdc17e42f..4d5217f5d63f5 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -3180,6 +3180,9 @@ def __init__( self.flags_with_values["enable_upsert_paged_spill"] = BOOLEAN_FLAG_VALUES self.flags_with_values["enable_upsert_chunked_stash"] = BOOLEAN_FLAG_VALUES self.flags_with_values["enable_upsert_async_reads"] = BOOLEAN_FLAG_VALUES + self.flags_with_values["enable_column_chunk_direct_compressed_output"] = ( + BOOLEAN_FLAG_VALUES + ) self.flags_with_values["column_chunk_compress_min_depth"] = [ "0", # compress every spilled body "1", # the default: fresh chunks store uncompressed diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index 01cf2f5292d9c..9ef2757cf3654 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -134,6 +134,15 @@ pub const COLUMN_CHUNK_COMPRESS_MIN_DEPTH: Config = Config::new( ParameterScope::Replica, ); +/// Bypass resident pool slots when committing compression-eligible chunk bodies. +pub const ENABLE_COLUMN_CHUNK_DIRECT_COMPRESSED_OUTPUT: Config = Config::new( + "enable_column_chunk_direct_compressed_output", + false, + "Compress spilled chunk output directly into extents without resident slot admission. \ + Applies at the compression depth floor and above, at each spill.", + ParameterScope::Replica, +); + /// Resident-bytes budget fraction for chunk spilling. Two consumers read /// it: the column pager's tiered policy multiplies it against the /// announced memory limit, and the buffer pool (`mz_ore::pool`) @@ -852,4 +861,5 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&COLUMN_PAGED_BATCHER_EAGER_BACKING) .add(&COLUMN_PAGED_BATCHER_POOL_RSS_TARGET_FRACTION) .add(&COLUMN_CHUNK_COMPRESS_MIN_DEPTH) + .add(&ENABLE_COLUMN_CHUNK_DIRECT_COMPRESSED_OUTPUT) } diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index b19c8d8044c65..600458a8b74ab 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -581,6 +581,9 @@ impl ComputeState { let compress_min_depth = u8::try_from(COLUMN_CHUNK_COMPRESS_MIN_DEPTH.get(config)).unwrap_or(u8::MAX); mz_timely_util::columnar::chunk::set_compress_min_depth(compress_min_depth); + mz_timely_util::columnar::chunk::set_direct_compressed_output( + ENABLE_COLUMN_CHUNK_DIRECT_COMPRESSED_OUTPUT.get(config), + ); } // Remember the maintenance interval locally to avoid reading it from the config set on diff --git a/src/ore/src/pool.rs b/src/ore/src/pool.rs index b8ac6d167af41..f0ab4ca01339d 100644 --- a/src/ore/src/pool.rs +++ b/src/ore/src/pool.rs @@ -201,6 +201,8 @@ pub struct PoolStats { pub inserts: u64, /// Inserts written directly to an extent because resident admission was full. pub direct_extent_inserts: u64, + /// Chunks inserted directly into an extent by caller request. + pub cold_inserts: u64, /// Reads submitted to the blocking executor. pub async_reads: u64, /// Submitted reads that have not released their concurrency permit. @@ -291,6 +293,7 @@ pub struct PoolStats { #[derive(Debug, Default)] struct Counters { direct_extent_inserts: AtomicU64, + cold_inserts: AtomicU64, async_reads: AtomicU64, inserts: AtomicU64, spill_scheduled: AtomicU64, @@ -717,6 +720,27 @@ impl Pool { ChunkHandle { meta } } + /// Encode `data` directly into an extent without allocating a resident slot. + /// + /// Compression is synchronous. The extent participates in the pool's + /// compressed-residency accounting and reclamation. Empty and oversize + /// payloads use the same fallback as [`Pool::insert_with`]. + pub fn insert_cold( + &self, + data: &[u64], + hints: ChunkHints, + codec: &'static dyn ExtentCodec, + ) -> ChunkHandle { + if data.is_empty() || region::size_class_for(std::mem::size_of_val(data)).is_none() { + return self.insert_with(data.len(), hints, codec, |dst| dst.copy_from_slice(data)); + } + let inner = &self.0; + let extent = SwapExtent::write(&inner.extent_arena, data, codec, Scratch::Shrink); + inner.counters.inserts.fetch_add(1, Ordering::Relaxed); + inner.counters.cold_inserts.fetch_add(1, Ordering::Relaxed); + self.finish_extent(data.len(), hints, codec, extent) + } + fn insert_extent( &self, len: usize, @@ -731,6 +755,21 @@ impl Pool { let inner = &self.0; let extent = SwapExtent::write(&inner.extent_arena, &words, codec, Scratch::Shrink); drop(words); + inner + .counters + .direct_extent_inserts + .fetch_add(1, Ordering::Relaxed); + self.finish_extent(len, hints, codec, extent) + } + + fn finish_extent( + &self, + len: usize, + hints: ChunkHints, + codec: &'static dyn ExtentCodec, + extent: SwapExtent, + ) -> ChunkHandle { + let inner = &self.0; let meta = Arc::new(ChunkMeta::new( inner, len, @@ -742,10 +781,6 @@ impl Pool { None, )); inner.live_chunks.fetch_add(1, Ordering::Relaxed); - inner - .counters - .direct_extent_inserts - .fetch_add(1, Ordering::Relaxed); { let mut state = meta.state(); inner.commit_extent(&meta, &mut state, extent); @@ -760,6 +795,7 @@ impl Pool { PoolStats { inserts: c.inserts.load(Ordering::Relaxed), direct_extent_inserts: c.direct_extent_inserts.load(Ordering::Relaxed), + cold_inserts: c.cold_inserts.load(Ordering::Relaxed), async_reads: c.async_reads.load(Ordering::Relaxed), async_reads_in_flight: { #[cfg(feature = "async")] @@ -3940,6 +3976,61 @@ mod tests { assert_eq!(range, want[8..24], "range reads copy the range directly"); } + #[mz_ore::test] + fn cold_insert_skips_resident_admission() { + let codecs: [&'static dyn ExtentCodec; 2] = [&TEST_CODEC, &IDENTITY_CODEC]; + for codec in codecs { + let pool = test_pool(256 << 20); + pool.set_rss_target(1 << 30); + let want = payload(SMALL, 703); + let handle = pool.insert_cold(&want, ChunkHints { depth: 3 }, codec); + assert_eq!(handle.residency(), Residency::Evicted); + assert_eq!(handle.meta.depth, 3); + let stats = pool.stats(); + assert_eq!(stats.inserts, 1); + assert_eq!(stats.cold_inserts, 1); + assert_eq!(stats.direct_extent_inserts, 0); + assert_eq!(stats.resident_bytes, 0); + assert_eq!(stats.live_chunks, 1); + assert!(stats.extent_resident_bytes > 0); + assert_eq!(read(&handle), want); + assert_eq!(pool.stats().resident_bytes, 0); + let mut range = Vec::new(); + handle.read_range_into(3..11, &mut range); + assert_eq!(range, want[3..11]); + assert_eq!(read_admit(&handle), want); + assert!(pool.stats().resident_bytes > 0); + pool.evict(&handle); + assert_eq!( + pool.stats().extent_bytes_written, + stats.extent_bytes_written + ); + assert_eq!(read(&handle), want); + drop(handle); + let stats = pool.stats(); + assert_eq!(stats.resident_bytes, 0); + assert_eq!(stats.live_chunks, 0); + assert_eq!(stats.extent_resident_bytes, 0); + assert_eq!(stats.frees, 1); + } + } + + #[mz_ore::test] + #[cfg_attr(miri, ignore)] + fn cold_insert_empty_and_oversize_fallbacks() { + let pool = test_pool(usize::MAX); + let empty = pool.insert_cold(&[], ChunkHints::default(), &TEST_CODEC); + assert_eq!(read(&empty), Vec::::new()); + let want = payload(SIZE_CLASSES[SIZE_CLASSES.len() - 1] / 8 + 1, 704); + let big = pool.insert_cold(&want, ChunkHints::default(), &TEST_CODEC); + assert_eq!(big.residency(), Residency::Oversize); + assert_eq!(read(&big), want); + assert_eq!(pool.stats().inserts, 2); + assert_eq!(pool.stats().cold_inserts, 0); + drop(big); + assert_eq!(pool.stats().resident_bytes, 0); + } + #[mz_ore::test] fn insert_with_fills_in_place() { let pool = test_pool(usize::MAX); diff --git a/src/storage/src/upsert_continual_feedback_v2.rs b/src/storage/src/upsert_continual_feedback_v2.rs index d00b818cc9379..f47c89f947112 100644 --- a/src/storage/src/upsert_continual_feedback_v2.rs +++ b/src/storage/src/upsert_continual_feedback_v2.rs @@ -1776,12 +1776,21 @@ mod test { #[mz_ore::test(tokio::test)] #[cfg_attr(miri, ignore)] async fn drain_reads_spilled_chunks() { + for direct in [false, true] { + assert_spilled_drain(direct); + } + } + + fn assert_spilled_drain(direct: bool) { use mz_ore::pool::Pool; - use mz_timely_util::columnar::chunk::set_spill_override; + use mz_timely_util::columnar::chunk::{ + set_direct_compressed_output_override, set_spill_override, + }; let pool = Pool::new().expect("pool creation"); pool.set_budget(0); set_spill_override(Some(pool.clone())); + set_direct_compressed_output_override(Some(direct)); const KEYS: i64 = 1500; let actual = upsert_test!(|input, persist, worker| { @@ -1793,6 +1802,10 @@ mod test { for k in 0..KEYS { input.send(((key(k), Some(Ok(row(k, k + 1))), 1), new_ts(1), Diff::ONE)); + if k % 500 == 499 { + input.flush(); + worker.step(); + } } input.advance_to(new_ts(2)); worker.step(); @@ -1801,6 +1814,8 @@ mod test { }); set_spill_override(None); + set_direct_compressed_output_override(None); + assert_eq!(pool.stats().cold_inserts > 0, direct); assert!( pool.stats().inserts > 0, "chunks should have spilled through the pool" diff --git a/src/timely-util/src/columnar/chunk.rs b/src/timely-util/src/columnar/chunk.rs index 4c39d35712f66..b5a031b08f740 100644 --- a/src/timely-util/src/columnar/chunk.rs +++ b/src/timely-util/src/columnar/chunk.rs @@ -39,9 +39,7 @@ //! actually touches. use std::borrow::Cow; -#[cfg(test)] -use std::cell::Cell; -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; use std::collections::VecDeque; use std::rc::Rc; use std::sync::Arc; @@ -84,7 +82,9 @@ thread_local! { #[cfg(test)] static COMPRESS_MIN_DEPTH_OVERRIDE: Cell> = const { Cell::new(None) }; - /// Reusable staging for call-scoped reads of spilled bodies. + static DIRECT_COMPRESSED_OUTPUT_OVERRIDE: Cell> = const { Cell::new(None) }; + + /// Reusable staging for call-scoped reads and serialization of spilled bodies. static READ_SCRATCH: RefCell> = const { RefCell::new(Vec::new()) }; } @@ -119,6 +119,28 @@ pub fn set_spill_override(pool: Option) { SPILL_OVERRIDE.with(|cell| *cell.borrow_mut() = pool); } +static DIRECT_COMPRESSED_OUTPUT: AtomicBool = AtomicBool::new(false); + +/// Compress eligible spilled bodies directly into extents, bypassing resident slots. +/// +/// Consulted at each spill. Bodies below the compression floor retain normal +/// admission. Encoding is synchronous, and short-lived bodies lose write elision. +pub fn set_direct_compressed_output(enabled: bool) { + DIRECT_COMPRESSED_OUTPUT.store(enabled, Ordering::Relaxed); +} + +/// Override direct compressed output on this thread for tests and benchmarks. +/// `None` restores the process setting. +pub fn set_direct_compressed_output_override(enabled: Option) { + DIRECT_COMPRESSED_OUTPUT_OVERRIDE.with(|cell| cell.set(enabled)); +} + +fn direct_compressed_output() -> bool { + DIRECT_COMPRESSED_OUTPUT_OVERRIDE + .with(|cell| cell.get()) + .unwrap_or_else(|| DIRECT_COMPRESSED_OUTPUT.load(Ordering::Relaxed)) +} + /// The youngest generational depth whose spilled bodies are compressed. See /// [`set_compress_min_depth`]. static COMPRESS_MIN_DEPTH: AtomicU8 = AtomicU8::new(DEFAULT_COMPRESS_MIN_DEPTH); @@ -188,12 +210,12 @@ fn spill_pool() -> Option { } } -/// Scratch capacity retained across reads, in words. A read larger than this +/// Scratch capacity retained across calls, in words. A read larger than this /// releases the buffer afterward, so a thread's scratch does not ratchet to /// the largest body it ever carried (heap no pool gauge can see). const SCRATCH_RETAIN_WORDS: usize = 1 << 18; -/// Run `f` with this thread's read scratch, cleared of any previous use. +/// Run `f` with this thread's scratch, cleared of any previous use. fn with_scratch(f: impl FnOnce(&mut Vec) -> Out) -> Out { READ_SCRATCH.with(|cell| { let mut scratch = cell.take(); @@ -447,7 +469,14 @@ impl ColumnChunk { let mut fences = D::Container::default(); fences.push(view.0.get(0)); fences.push(view.0.get(records - 1)); - let handle = spill_column(column, pool, len_bytes, ChunkHints { depth }, codec); + let handle = spill_column( + column, + pool, + len_bytes, + ChunkHints { depth }, + codec, + compressed && direct_compressed_output(), + ); ColumnChunk::Spilled( Rc::new(SpilledBody { records, @@ -590,23 +619,36 @@ impl ExtentCodec for Lz4Codec { } } -/// Serialize a column into a pool slot. The `Align` variant is already the -/// serialized form and copies in directly. Other variants write their -/// [`ContainerBytes`] encoding through a cursor over the slot memory. Sizing -/// is exact, so a short or overlong write is a contract violation and panics. +/// Serialize a column into the pool, optionally bypassing resident admission. fn spill_column( column: Column, pool: &Pool, len_bytes: usize, hints: ChunkHints, codec: &'static dyn ExtentCodec, + direct: bool, ) -> ChunkHandle { mz_ore::soft_assert_eq_no_log!(len_bytes % 8, 0); + if direct { + return match column { + Column::Align(words) => pool.insert_cold(&words, hints, codec), + other => with_scratch(|words| { + words.resize(len_bytes / 8, 0); + serialize_column(other, words); + pool.insert_cold(words, hints, codec) + }), + }; + } + pool.insert_with(len_bytes / 8, hints, codec, |dst| { + serialize_column(column, dst) + }) +} + +fn serialize_column(column: Column, dst: &mut [u64]) { match column { - Column::Align(words) => { - pool.insert_with(words.len(), hints, codec, |dst| dst.copy_from_slice(&words)) - } - other => pool.insert_with(len_bytes / 8, hints, codec, |dst| { + Column::Align(words) => dst.copy_from_slice(&words), + other => { + let len_bytes = std::mem::size_of_val(dst); let bytes: &mut [u8] = bytemuck::cast_slice_mut(dst); let mut cursor = std::io::Cursor::new(bytes); other.into_bytes(&mut cursor); @@ -615,7 +657,7 @@ fn spill_column( len_bytes, "serialized body must fill the chunk exactly", ); - }), + } } } @@ -2420,6 +2462,40 @@ mod tests { set_compress_min_depth_override(None); } + #[mz_ore::test] + fn direct_output_respects_floor_and_preserves_serialization() { + let data: Vec = (0..2048u64).map(|k| ((k, k + 1), k % 3, 1)).collect(); + set_compress_min_depth_override(Some(1)); + for enabled in [false, true] { + set_direct_compressed_output_override(Some(enabled)); + for depth in [0, 1] { + for aligned in [false, true] { + let pool = Pool::new().expect("pool creation"); + pool.set_budget(usize::MAX); + let column = build_column(&data); + let mut expected = vec![0; column.length_in_bytes() / 8]; + serialize_column(column.clone(), &mut expected); + let column = if aligned { + Column::Align(expected.clone()) + } else { + column + }; + let chunk = TestChunk::spill_body(column, &pool, depth); + assert_eq!(pool.stats().cold_inserts, u64::from(enabled && depth >= 1)); + assert_eq!(pool.stats().resident_bytes == 0, enabled && depth >= 1); + let Column::Align(actual) = chunk.into_column() else { + panic!("a spilled body must read back serialized"); + }; + assert_eq!(actual, expected); + assert_eq!(pool.stats().resident_bytes, 0); + } + } + } + set_direct_compressed_output_override(None); + set_compress_min_depth_override(None); + READ_SCRATCH.with(|scratch| assert!(scratch.borrow().capacity() <= SCRATCH_RETAIN_WORDS)); + } + /// Re-spilling an already-serialized body exercises the `Column::Align` /// branch of `spill_column` and round-trips byte-identically. #[mz_ore::test] diff --git a/src/timely-util/src/pool_config/metrics.rs b/src/timely-util/src/pool_config/metrics.rs index 9d7d32b4f5049..4f2d6c291f196 100644 --- a/src/timely-util/src/pool_config/metrics.rs +++ b/src/timely-util/src/pool_config/metrics.rs @@ -41,6 +41,7 @@ pub fn register(registry: &MetricsRegistry) { // monotonic. gauge(registry, metric!(name: "mz_column_pool_async_reads_total", help: "Pool reads submitted to the blocking executor."), |s| s.async_reads); gauge(registry, metric!(name: "mz_column_pool_async_reads_in_flight", help: "Submitted pool reads that have not released their concurrency permit."), |s| s.async_reads_in_flight); + gauge(registry, metric!(name: "mz_column_pool_cold_inserts_total", help: "Chunks inserted directly into an extent by caller request."), |s| s.cold_inserts); gauge(registry, metric!(name: "mz_column_pool_direct_extent_inserts_total", help: "Inserts written directly to an extent because resident admission was full."), |s| s.direct_extent_inserts); gauge(registry, metric!(name: "mz_column_pool_resident_bytes", help: "Uncompressed bytes resident in the buffer pool."), |s| s.resident_bytes); gauge(registry, metric!(name: "mz_column_pool_oversize_bytes", help: "Bytes held by oversize chunks that bypass pool paging."), |s| s.oversize_bytes); From 4b2082723270c418e77e7615ccd70f6367fbdaff Mon Sep 17 00:00:00 2001 From: Dov Alperin Date: Tue, 8 Sep 2026 21:58:41 -0400 Subject: [PATCH 05/31] storage: prototype payload-separated upsert for hydration benchmarks --- .../design/20260908_out_of_core_operators.md | 572 ++++++++++++++++++ misc/python/materialize/mzcompose/__init__.py | 5 + .../materialize/parallel_workload/action.py | 1 + src/storage-types/src/dyncfgs.rs | 10 + .../src/upsert_continual_feedback_v2.rs | 322 +++++++++- .../upsert_continual_feedback_v2/payload.rs | 448 ++++++++++++++ src/timely-util/src/columnar.rs | 1 + src/timely-util/src/columnar/chunk.rs | 10 +- src/timely-util/src/columnar/payload.rs | 392 ++++++++++++ src/timely-util/src/lib.rs | 1 + src/timely-util/src/out_of_core.rs | 34 ++ src/timely-util/src/out_of_core/batch.rs | 66 ++ src/timely-util/src/out_of_core/operators.rs | 181 ++++++ src/timely-util/src/out_of_core/payload.rs | 568 +++++++++++++++++ src/timely-util/src/out_of_core/tests.rs | 515 ++++++++++++++++ .../mzcompose.py | 1 + 16 files changed, 3095 insertions(+), 32 deletions(-) create mode 100644 doc/developer/design/20260908_out_of_core_operators.md create mode 100644 src/storage/src/upsert_continual_feedback_v2/payload.rs create mode 100644 src/timely-util/src/columnar/payload.rs create mode 100644 src/timely-util/src/out_of_core.rs create mode 100644 src/timely-util/src/out_of_core/batch.rs create mode 100644 src/timely-util/src/out_of_core/operators.rs create mode 100644 src/timely-util/src/out_of_core/payload.rs create mode 100644 src/timely-util/src/out_of_core/tests.rs diff --git a/doc/developer/design/20260908_out_of_core_operators.md b/doc/developer/design/20260908_out_of_core_operators.md new file mode 100644 index 0000000000000..3c91c8c658852 --- /dev/null +++ b/doc/developer/design/20260908_out_of_core_operators.md @@ -0,0 +1,572 @@ +# A shared model for out-of-core operators + +- Status: discussion draft, September 8, 2026. Interfaces and rollout gates below are proposals. +- Related: [Buffer-managed dataflow state](20260610_buffer_managed_state.md). +- Related experiment: [upsert hydration optimizations, PR #38719](https://github.com/MaterializeInc/materialize/pull/38719). + +## Proposal in brief + +Separate operator state into immutable payload storage and a spillable index of +keys, handles, timestamps, and differences. Execute operators through resumable, +budgeted work units that request payloads only when their semantics require them. +Reuse Differential's time and difference machinery where possible, including the +boundary demonstrated by `int_proxy`. + +Upsert is the first consumer. Equijoin is the second design and implementation +check. Neither storage nor scheduling should know about source offsets, +latest-value selection, join predicates, or a particular reduction function. + +The central hypothesis is that moving compact metadata through repeated merges, +while keeping payload blocks independently owned, reduces byte amplification. +Whether this improves elapsed time depends on the additional lookup, equality, +fragmentation, and payload-read costs. We will measure those costs separately. + +## The problem + +Our chunked state paths can repeatedly serialize, copy, compress, and decompress +wide rows while sorting or merging updates. Much of that work is needed to move +state through the representation, even when the operation only needs a key, +timestamp, or source ordering field to make its decision. + +The existing pool controls chunk residency. It does not by itself separate payload +lifetime from merge lifetime. Smaller chunks, tighter admission, direct compressed +output, and offloaded reads improve the current representation. A shared operator +model should also let a merge keep a payload reference instead of rebuilding the +payload at every generation. + +This is broader than upsert hydration. Joins need to find matching groups and +materialize selected pairs. Reductions need to replay histories and reconcile +outputs. All need progress tracking, bounded staging, storage ownership, and an +execution path that can wait for unavailable data without blocking a worker. + +## Success criteria + +- Upsert and equijoin share payload storage, ownership, memory accounting, and + read scheduling. Their semantic rules remain in their operator implementations. +- A metadata merge can retain live payloads without decoding or rewriting them. + Repacking payload blocks is separately scheduled and measured. +- Payloads, indexes, handle translation, ownership metadata, staging, and queued + work have explicit memory charges. Increasing state beyond the budget does not + introduce an unbounded resident table of row handles or block descriptors. +- Supported operators make progress under a small budget, including with skew, + slow reads, output backpressure, cancellation, and concurrent consumers. +- Results and progress agree with existing operators under arbitrary valid + batching, retractions, timestamp order, compaction, and restarts. +- Deep-state workloads improve without an unacceptable resident-state penalty. + Proposed acceptance thresholds appear under measurement, rather than assuming + that fewer copied bytes guarantee faster queries. + +## Scope + +This proposal covers recreatable local operator state. Persist remains the durable +source of recovery. Local handles do not become part of persisted records or the +wire protocol. Crossing a process boundary materializes or explicitly transfers +owned data rather than sending a process-local handle. + +We will preserve existing in-memory implementations during evaluation. Converting +all operators, implementing a new durable store, and selecting a particular kernel +async I/O API are outside the first delivery. General reduction remains a design +requirement, but arbitrary reduction callbacks do not automatically acquire a +bounded-memory implementation through this interface. + +## What int-proxy contributes + +Differential's `int_proxy` tactics exchange consolidated presentations of: + +```text +((key_hash: u64, value_id: u64), time, diff) +``` + +The tactic performs the time and difference computation. The backend interprets +values and constructs outputs. A join returns matched IDs with joined times and +multiplied differences. A reduce asks the backend for corrections at the times +that require reconciliation. + +The two integers have different contracts. `key_hash` partitions independent +work, but collisions are allowed. `value_id` identifies data within the backend's +presentation. Distinct logical data in the same hash group must remain distinct, +and equal data must be presented consistently wherever cancellation is required. +Input and output presentations can have separate ID namespaces. + +The local reference reduce backend assigns IDs per window and resolves them +through row vectors. It still clones rows and emits ordinary row batches. The +proxy interface therefore demonstrates a separation of responsibilities, not a +ready-made payload store or an out-of-core guarantee. + +Two constraints matter for this design: + +1. The callbacks are synchronous. A cold read cannot simply become an `await` + inside the existing tactic protocol. +2. The reduce window contract requires a complete key-hash group in the window + that first reports it. A window target does not bound a single enormous group. + Its novel time support must also survive consolidation: even an update that + cancels against history can introduce a time requiring reconciliation. + +Reference code is in the sibling Differential repository under +`differential-dataflow/src/operators/int_proxy/{mod,reduce,join,vec_backend}.rs`. +This draft describes that local implementation. Materialize currently locks +Differential 0.25.1, whose proxy API must be reconciled with the local work before +integration. Names and signatures here are illustrative, not a compatibility claim. + +## Architecture and ownership boundaries + +```mermaid +flowchart TD + A[Operator semantics: upsert, join, reduce] --> B[Resumable execution and time/diff logic] + B --> C[Spillable indexes and immutable batches] + B --> D[Budgeted payload reads] + C --> E[Payload store and ownership manifests] + D --> E + E --> F[Pool and extent storage] + G[Shared memory and I/O admission] -.-> B + G -.-> C + G -.-> D + G -.-> E +``` + +### Payload storage + +Store immutable row blocks independently of index batches. A block includes its +record boundaries and encoding information. The store supports batched resolution +of row handles, allowing requests to be grouped by block and decoded together. +The first implementation should reuse the pool's codecs and extent machinery. + +Reads initially return owned, budgeted decoded buffers. References into a decoded +buffer are scoped to a read lease. This follows the pool's current copy-out model +and avoids requiring a redesign around borrowed pointers into evictable slots. +A lease retains both the bytes' memory charge and ownership of the source needed +by an unfinished read. Callers cannot retain a naked row reference after the +lease expires. + +Block size is a policy choice, separate from index chunk size. Smaller blocks +reduce sparse-read amplification but add metadata and can weaken compression. +Benchmark several sizes rather than baking the upsert workload's preferred size +into the interface. Oversized individual rows require a charged exceptional path +or streaming support, never an unaccounted allocation. + +### Identity: handles need not be integers + +Use distinct types for storage identity and presentation identity: + +| Identity | Meaning | Lifetime | +| --- | --- | --- | +| `RowHandle` | Locates a stored row within a store namespace | While owning batches or read operations retain it | +| `ProxyId` | Identifies semantic data in an operator presentation | Until that presentation's work and output translation finish | +| Group key | Defines independent semantic work | Defined by the operator and its index | + +A candidate `RowHandle` representation is `(block_id, row_slot, generation)`. +Moving a block between resident memory and an extent changes its location, not +its identity. Generation checks prevent stale handles from resolving to reused +storage. The store namespace is supplied by the owning batch or execution context. + +The current proxy bridge requires `u64`, but the design does not require every +identity to be a `u64`. A backend can translate storage handles to dense integer +IDs for a bounded window. We should keep that adapter before generalizing the +upstream tactic's types unless two consumers demonstrate a need for the latter. +A handle's numeric order does not imply row order. + +Physical identity is also not semantic equality. Two independently ingested equal +rows can have different handles. Exact consolidation needs an equality mechanism: +for example, a fingerprint to find candidates followed by byte comparison, with +consistent proxy IDs assigned to equal data during presentation. Fingerprints +alone are insufficient. Such lookup state must be budgeted and spillable too. + +Do not require global interning of all live rows. Window-local canonicalization +and merge-time equality resolution are the starting point. Operators can exploit +stronger local knowledge, such as an already-owned output being reused, without +making that knowledge a storage requirement. + +### Indexes and batches + +An index entry carries the fields required for navigation and operator decisions, +plus references to payloads. Field projection belongs to an index layout or +operator backend. The storage layer does not assume a fixed tuple containing +upsert-specific ordering information. + +Keys can themselves be wide. A hash or compact prefix may identify candidate +groups, but exact key comparison can require payload reads. The execution protocol +must permit these reads during navigation and consolidation, not just at final +output. Likewise, a reducer that requires value ordering cannot sort opaque +handles and assume the order matches values. + +Index chunks, fences, equality indexes, and manifests must support external +storage. Resident roots and caches have byte budgets. Existing merge machinery +can be reused where it operates on the chosen compact representation. Comparator +or equality paths requiring cold data need an explicit suspend/resume boundary. + +### Ownership and reclamation + +A `RowHandle` is a locator, not an owning reference. An immutable batch manifest +owns the payload blocks its entries reference. Active reads and unpublished output +builders acquire ownership too. Prefer ownership per block or segment over an +atomic reference count for every row copy. + +Publishing an output batch must establish its manifest's ownership before input +ownership is released. Cancellation discards unpublished output and releases its +ownership. Dropping a batch releases references incrementally, with pending +release work charged and able to yield. A block is reclaimable only when no +published batch, builder, or active read owns it. A timestamp frontier alone is +not a reclamation proof. + +Coarse ownership can retain mostly dead blocks. Repacking live rows is a separate, +budgeted maintenance operation that creates new blocks and rewrites affected index +references. Old batches and active readers continue to own old blocks until they +retire. Do not introduce an unbounded per-row forwarding map to hide relocation. +Measure the temporary double ownership and charge the rewrite's scratch space. + +The ownership manifest and block directory are substantial parts of the work. +Retaining one always-resident `Arc` per live block merely moves the +scaling limit. A production design needs paged directory/ownership metadata, +bounded resident roots, and a way to retire metadata without loading an entire +batch manifest. The current pool API may need extensions at this boundary. + +## Execution, progress, and budgets + +The execution unit is a continuation with explicit input ownership, progress +holds, and a bounded working set. Its conceptual protocol is: + +```text +advance(work_budget) + -> need_reads(read_set, continuation) + | produced(output_batch, continuation) + | yield(continuation) + | complete + | error +``` + +These are protocol outcomes, not proposed Rust signatures. Metadata and payload +reads use the same admission rules. A ready-data step is synchronous and bounded +by bytes or work, with a separate fairness limit so a long CPU-only step yields. +Output backpressure suspends execution without accumulating unlimited ready output. + +The driver reserves decode/output memory and I/O capacity before issuing a read. +Completed buffers remain charged until consumed, including while the worker is +busy elsewhere. A limit on running reads alone is insufficient. Several operators +share process-level admission, with per-consumer fairness and an allowance for +work that releases memory. An operator must not hold the whole budget while +waiting for an additional allocation needed to make progress. The prototype must +exercise this deadlock case and establish a reservation/release discipline. + +No pool locks, borrowed cursors into mutable state, or uncharged buffers may cross +a suspension. Cancellation retains a running job's lease and permit until that +job actually finishes, then releases them even if its consumer disappeared. +Storage errors invalidate the work and are surfaced through the operator's error +or restart path. Partially built output is not published as a completed batch. + +Suspension must preserve Differential's time semantics. Work retains capabilities +or equivalent holds for every timestamp at which it can still emit. A yield or +pending read must not advance the output frontier. Publication and continuation +updates must avoid duplicate output when work resumes. Partial-order timestamps, +compaction, and novel time support remain the responsibility of the time/diff +machinery and its adapter, rather than the payload store. + +Time/diff bookkeeping is part of the budget too. Replay histories, pending-time +schedules, and retained presentations must be spillable or have an explicit +admission bound. The tactic's existing resident pending-time map cannot be treated +as free metadata. Any bound must preserve required times and holds, using +backpressure or external state rather than dropping work. This is a dependency of +the resumable tactic design, not something the payload store can solve alone. + +There are two integration options: prepare a bounded, fully resident presentation +before invoking a synchronous tactic, or make tactic execution resumable. The +first is useful for a prototype, but cannot provide the complete contract for +large groups or payload-dependent comparisons. The target design requires a +resumable path. We should upstream the smallest suitable execution boundary +rather than duplicate Differential's time logic in each backend. + +The initial I/O implementation can offload swap faults and decoding to a bounded +blocking executor. It does not guarantee fault-free worker execution, and it is +not native asynchronous file I/O. The same request/continuation contract should +permit a later file-extent implementation without changing operator semantics. + +## Large groups and operator contracts + +A hot key can exceed any window budget. Different operators need different +strategies, and declaring an arbitrary split into independent windows is incorrect. + +| Consumer | Small representation | Payload access | Large-group strategy | +| --- | --- | --- | --- | +| Upsert | Key, time, source order, optional row handle | Equality and old/new rows needed for output | Stream latest-update selection while preserving the existing feedback and frontier rules | +| Equijoin | Key projection/hash, row handles, time, diff | Exact key checks, predicates, output construction | Block both sides, retain resumable history, and stream matches under output backpressure | +| Threshold or aggregatable reduce | Group identity, value identity or sufficient aggregate, time, diff | Depends on equality and aggregate semantics | Use valid partial state or spillable histories | +| General reduce | Group and value references with histories | Potentially all values in semantic order | Require a streaming callback or external intermediate state; an arbitrary slice callback has no bounded-memory guarantee | + +Upsert's source stash can select by source order without reading every row. +Its feedback state still requires exact row identity for retractions and +consolidation. Equality reconciliation after persist readback must not assume +that a deserialized row preserves its former local handle. + +Equijoin must enumerate all valid matches even when output greatly exceeds input. +That cost cannot be removed by proxies. Blocking may require rereading one side, +so we must measure actual read amplification and scheduler fairness. + +These two consumers should use the same lower layers. If implementing equijoin +requires a second payload store or ownership mechanism, the proposed abstraction +has not met its purpose. + +## Minimal viable prototype and delivery + +This document precedes a prototype so we can agree on the ownership and execution +boundaries before implementing them. No performance gains from this representation +have been measured yet. + +1. Build a small shared store/index experiment with a deliberately tiny budget, + duplicate logical rows at different handles, asynchronous reads, and cancellation. + Exercise both an upsert-like selector and an equijoin over it. Keep the interfaces + provisional until both work. A fully resident metadata directory can be used to + establish mechanics, but cannot satisfy the out-of-core acceptance gate. +2. Implement spillable metadata and manifests, exact equality reconciliation, + reclamation, large-group continuations, and publication/progress invariants. + Demonstrate budget and liveness properties independently of hydration speed. +3. Integrate upsert while preserving its existing time and feedback semantics. + Compare against v1 and the measured optimization stack. Keep persisted and wire + representations unchanged so a fresh replica can select either implementation. +4. Integrate equijoin using the same store, admission, and ownership APIs. Validate + skew and output backpressure. Use a reduction consumer to settle remaining + value-order and large-group interface questions before claiming broad support. +5. Roll out per consumer behind flags. Production defaults off, test/CI defaults + on for supported paths. Check small-state overhead and deep-state behavior before + increasing exposure. Rollback recreates local state from persist rather than + converting live handles between formats. + +## Correctness and measurement + +Correctness testing compares output updates and progress with existing operators, +including product timestamps with incomparable elements. Vary input batch order, +negative differences, frontier advancement, compaction, empty inputs, deletion, +and duplicate rows. Force hash collisions and separate physical copies of equal +data. Preserve the case where novel input cancels against history but still +introduces a time requiring reconciliation. + +Storage and scheduler tests cover eviction during reads, cancellation before and +after submission, delayed completion, stale handle generations, output publication +before input release, repacking while old batches remain live, scratch exhaustion, +and injected read failures. Run with budgets smaller than a group and with two +competing consumers. Verify both progress and eventual release of charges. +Restart tests reconstruct state from persist and verify that no local handles leak +into durable or exchanged data. + +Use the spec-sheet persisted-seed workflow for upsert hydration: load once, stop +the writer, and recreate a replica for each variant. Record the exact image, +configuration, seed shape, worker count, CPU, memory, and usable scratch/swap. +Compare the current optimized representation with the proposed one on the same +build where possible. Keep a v1 control, but do not make v1 parity the only success +criterion for a shared framework. + +Measure equijoin separately with fixed input/output cardinalities, selective and +dense matches, skew, and wide join keys. Add steady-state update/retraction runs +and concurrent consumers. The first upsert screening replica has one worker and +half a CPU; it cannot establish production concurrency or scheduling behavior. + +Sweep payload width, actual compressibility, state-to-budget ratio, update density, +key skew, worker/core ratio, and payload block size in stages rather than running +an undirected Cartesian product. Include resident-state controls and poorly +compressible data. Randomize or counterbalance variant order, retain failed runs, +and repeat enough trials to report variation rather than one timing. + +Collect elapsed time, CPU, worker scheduling delay, time waiting for reads, decode +time, peak RSS, swap, and all memory-ledger components. Count metadata merge bytes, +payload bytes copied/decoded/rewritten, equality lookups and verification reads, +block cache hits, dead-byte retention, reclamation work, and output bytes. Separate +extent creation from physical device traffic. Account for sampling gaps and missing +metrics explicitly. Treat restarts as failed trials, not successful slow runs. + +Proposed gates for discussion: + +- Correctness and bounded-memory/liveness checks pass for both initial consumers. +- Supported deep-state workloads finish under a fixed budget without growing an + unaccounted directory, pending-read queue, or output queue. +- On wide-row, deep-state cases with comparable output, reduce payload bytes + processed by state maintenance by at least 2x versus the optimized chunk path, + and demonstrate an elapsed-time improvement outside run-to-run variation. +- Target no more than 5% resident-state slowdown for each consumer. If overhead + exceeds that threshold, investigate layout/admission costs or keep an explicit + small-state path. These thresholds are proposed acceptance criteria, not forecasts. + +## Local mechanics prototype + +The prototype lives in [`mz_timely_util::out_of_core`](../../../src/timely-util/src/out_of_core.rs). +It provides a shared payload store, immutable block manifests, resident metadata +batches, and whole-step read admission. A read request owns its blocks before it +can suspend. A blocking read job owns its byte and job permits through +cancellation, and the returned lease retains the byte reservation until its +borrowed output is consumed. + +Two consumers share these interfaces: maximum-order selection and equijoin. +Selection rebuilds metadata while reusing existing payload blocks. The resumable +join cursor emits one pair per step and uses the same store to acquire both +payloads atomically. String and composite keys work without integer encoding. + +The standalone proxy adapters and live Differential harness are retained in the +local prototype workspace. The benchmark image includes the shared storage and +native columnar upsert integration below, using Materialize's existing Timely +and Differential dependencies. It has no sibling-checkout dependency. + +## Live dataflow setup + +`out_of_core::live` in the standalone prototype workspace installs +operators into a running Timely dataflow using the local Differential checkout: + +1. `store_payloads` exchanges serialized input rows by key, then packs their + bytes into the worker's payload store. It emits `Stored` values whose + ordering and equality use a caller-defined logical identity. Each value owns + its payload block. Handles never cross workers. +2. `arrange_payloads` builds a Differential chunk spine over key hashes, exact + keys, and stored values. Trace consolidation and merging clone metadata and + ownership references, without copying payload bytes. +3. `latest` runs `ProxyReduceTactic` through `reduce_with_tactic`. `join` runs + `ProxyJoinTactic` through `join_with_tactic`. These drivers maintain input and + output history, compaction frontiers, and capabilities across successive + input batches. The backends resolve exact identities and hash collisions. +4. `fetch` requests the complete read set for an output record, polls its future + on the Timely worker, and uses a synchronous activator to resume when I/O + completes. Blocking pool reads run through the Tokio executor. Each future + holds an output capability and its stored values until publication. Incoming + antichain stamps are retained as capability sets before deriving a capability + at each record's time. + +The integration harness runs selection and join together on two workers and +compares both outputs against standard Differential operators. Inputs advance +through multiple epochs and are probed before sending subsequent epochs. It also +covers product timestamps, two-element message stamps, error propagation, a +blocked read that leaves another dataflow schedulable, and dropping a dataflow +while its blocking read still owns admission. + +```sh +MZ_DEV_BUILD_SHA=f0e632d1 cargo +1.97.1 nextest run \ + -p mz-timely-util --features out-of-core-prototype --test out_of_core_live + +MZ_OOC_EPOCHS=1000 MZ_DEV_BUILD_SHA=f0e632d1 cargo +1.97.1 nextest run \ + -p mz-timely-util --features out-of-core-prototype --test out_of_core_live \ + -E 'test(live_proxy_pipeline_matches_reference_on_two_workers)' --no-capture +``` + +A local 1,000-epoch run matched both reference outputs. With a 1,024-byte decoded +budget per worker, observed peaks were 1,024 and 992 bytes. Live payload blocks +sampled after each completed epoch peaked at 10 and 5, and both workers released +all blocks on closure. This is a functional small-state run over compressed +extents, not a hydration or disk-throughput benchmark. + +The generic live harness remains separate from Materialize's source and compute +renderers. The native upsert integration below uses the same storage and ownership +model with Materialize's current Differential dependency and existing feedback +protocol. + +The remaining shared runtime work includes spillable metadata and ownership +indexes, exact row equality where the caller has no logical identity, resumable +window presentation and history loading, decoded-block reuse, and admission covering +queued metadata and output. Production wiring also needs one compatible Timely +and Differential dependency set, row/error codecs, operator shutdown integration, +metrics, configuration defaults, and restart/rehydration validation. + +### Resumable join matching + +The local proxy join retains direct-cross positions or bilinear replay histories +across calls. Each prepared work unit clones its configured backend, giving it a +private ID interpretation table. A window can produce multiple `cross` calls, +so that table remains valid until the next window or work-unit drop. + +`JoinWork::step` returns output, yield, or done. The live driver retains its +capabilities on yield and reactivates the operator, including when exact-key +filtering discards every candidate. Matching defaults to 4,096 candidate matches +and 4,096 replay transitions per quantum. The compatibility iterator consumes +yields internally and provides no scheduling guarantee. + +Window presentation, history loading, and consolidation remain synchronous and +can exceed a quantum. The limits bound staged matches and matching transitions, +not the whole activation, input memory, or arbitrary backend output expansion. +A live test checks a 10,000-pair hot key through pool-backed payload fetches. + + +### Native columnar upsert integration + +`enable_upsert_payload_stash` selects a third upsert-v2 state representation. It +is off in production and on in the mzcompose test parameter defaults. It takes +precedence over `enable_upsert_chunked_stash` when upsert-v2 is enabled. + +`columnar::payload::PayloadChunk` wraps a normal columnar metadata chunk and a +manifest. The existing Differential chunk batcher and spine handle merging, +advancement, sealing, and compaction. Metadata can spill through the existing +columnar path. Each rewrite retains exactly the referenced payload blocks, +without decoding payloads. The generic bulk-probe interface returns metadata +with the ownership needed to resolve its locators. + +The source stash stores keys, times, source offsets, tombstones, and row handles. +Offset selection is the existing `UpsertDiff` semigroup. Payloads are published +after the initial chunker selects its winners, in metadata order. Ineligible updates retain +their handles when re-stashed. The feedback arrangement stores keys and exact +payload identities with ordinary additive diffs. An operator-local weak equality +index uses fingerprints to find candidates, then checks their bytes before +reusing a live handle. It owns no blocks and sweeps dead entries incrementally. +This identity policy is specific to this consumer, rather than required by the +chunk abstraction. Its resident index cost needs measurement. + +Source and feedback share one payload store: 2 MiB blocks, a 16 MiB decoded-read +budget, and two blocking read jobs. The drain retains at most two decoded blocks +for reuse across windows. Each 1,024-record window groups emitted updates by +payload locator before decoding, avoiding old/new block alternation. Feedback +canonicalization retains one decoded block. Payloads larger +than a block remain inline, preserving supported row sizes at the cost of the +original row-copying behavior for those rows. Encoded error rows follow the same +path as successful values. The persist format is unchanged. + +The upsert driver still owns resume filtering, persist eligibility, frontiers, +metrics, and shutdown. This does not replace that protocol with a generic latest +reduction, and does not install the local proxy join driver into compute. It +requires no Timely or Differential dependency migration. + +Limits to measure include the resident equality index and manifests, metadata +reads while pruning ownership, synchronous metadata merges, batching scratch, +and payload block retention when only a few rows remain live. Decoded admission +is shared by the source and feedback operators of one upsert dataflow, not yet +across every operator in the process. The process pool still governs residency. + +The hydration comparison adds `v2_payload` alongside `v1` and `v2_all`, using +identical persisted state and fresh replicas. The candidate image must contain +the new flag. The existing overnight image predates this integration. + + +## Alternatives + +**Continue improving combined row chunks.** This has the smallest implementation +cost and preserves write elision for short-lived chunks. It remains the baseline. +It is sufficient if measurements show that byte amplification is no longer the +limiting cost, and avoids the proposed equality and ownership overhead. + +**Adopt the existing int-proxy reference backend directly.** This validates tactic +integration, but does not eliminate row copying or provide bounded cold reads, +metadata, or large-group execution. It is a useful reference, not the target store. + +**Globally intern every row to an integer.** Equality becomes cheap after lookup, +but the interning index and ownership can dominate memory and I/O. Requiring it +also couples every consumer to a global identity policy. Prefer bounded exact +canonicalization, with stronger interning optional when measurements justify it. + +**Use one key/value store for all operator state.** This can supply storage and +lookup, but still needs integration with immutable batch sharing, time histories, +progress, and output backpressure. It may be a backend option rather than the +operator interface itself. + +**Make native async I/O the first project.** It can improve read scheduling, but +preserves unnecessary bytes read, written, and decoded. Separate the operator's +resumable contract from the extent implementation so both improvements can be +measured independently. + +## Questions for discussion + +1. **Execution boundary:** should Differential expose resumable tactics, or a + smaller replay primitive that a Materialize async driver composes? Preparing + resident windows is a useful first experiment, not a complete large-group answer. +2. **Ownership granularity:** can block/segment manifests and paged ownership + metadata provide acceptable dead-byte retention, or do we need finer liveness + information? What is the minimum pool API extension required? +3. **Equality:** which consumers can canonicalize within a merge or work window, + and which need a longer-lived spillable equality index? How much cold comparison + traffic does each approach introduce? +4. **Large groups:** which reduction contracts are worth supporting initially, + and how should operators declare their streaming or external-state requirements? +5. **Delivery gate:** do we agree that equijoin must exercise the shared runtime + before these APIs are considered stable, even though upsert ships first? +6. **Performance gate:** are the proposed 2x byte-reduction and 5% resident-overhead + targets appropriate, and which workloads should decide whether the additional + storage machinery is justified? diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index f57c5ea6d1789..de91f84b6a89e 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -369,6 +369,11 @@ def get_variable_system_parameters( "true", ["true", "false"], ), + VariableSystemParameter( + "enable_upsert_payload_stash", + "true", + ["true", "false"], + ), VariableSystemParameter( "enable_upsert_async_reads", "true", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 4d5217f5d63f5..7dfdac8e99ccd 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -3180,6 +3180,7 @@ def __init__( self.flags_with_values["enable_upsert_paged_spill"] = BOOLEAN_FLAG_VALUES self.flags_with_values["enable_upsert_chunked_stash"] = BOOLEAN_FLAG_VALUES self.flags_with_values["enable_upsert_async_reads"] = BOOLEAN_FLAG_VALUES + self.flags_with_values["enable_upsert_payload_stash"] = BOOLEAN_FLAG_VALUES self.flags_with_values["enable_column_chunk_direct_compressed_output"] = ( BOOLEAN_FLAG_VALUES ) diff --git a/src/storage-types/src/dyncfgs.rs b/src/storage-types/src/dyncfgs.rs index e27dff344b92e..066e8f53b1bf9 100644 --- a/src/storage-types/src/dyncfgs.rs +++ b/src/storage-types/src/dyncfgs.rs @@ -452,6 +452,15 @@ pub const ENABLE_UPSERT_ASYNC_READS: Config = Config::new( ParameterScope::Replica, ); +/// Separate upsert-v2 payload blocks from columnar merge metadata. +/// Read once per dataflow. Uses the process pool and asynchronous payload reads. +pub const ENABLE_UPSERT_PAYLOAD_STASH: Config = Config::new( + "enable_upsert_payload_stash", + false, + "Use payload-separated columnar state for upsert-v2. Takes effect on new dataflows.", + ParameterScope::Replica, +); + // RocksDB /// How many times to try to cleanup old RocksDB DB's on disk before giving up. @@ -577,6 +586,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&ENABLE_UPSERT_PAGED_SPILL) .add(&ENABLE_UPSERT_CHUNKED_STASH) .add(&ENABLE_UPSERT_ASYNC_READS) + .add(&ENABLE_UPSERT_PAYLOAD_STASH) .add(&WALLCLOCK_GLOBAL_LAG_HISTOGRAM_RETENTION_INTERVAL) .add(&WALLCLOCK_LAG_HISTORY_RETENTION_INTERVAL) .add(&crate::sources::sql_server::CDC_CLEANUP_CHANGE_TABLE) diff --git a/src/storage/src/upsert_continual_feedback_v2.rs b/src/storage/src/upsert_continual_feedback_v2.rs index f47c89f947112..7551cae9482b3 100644 --- a/src/storage/src/upsert_continual_feedback_v2.rs +++ b/src/storage/src/upsert_continual_feedback_v2.rs @@ -64,9 +64,12 @@ //! ## Stash flavors //! //! [`UpsertStashFlavor`], resolved from `enable_upsert_chunked_stash` at -//! operator construction, selects between two instantiations of the same +//! operator construction, selects between three instantiations of the same //! loop: //! +//! * **Payload**: columnar chunks contain metadata and payload locators. +//! Separate manifests own pool-backed rows, and feedback uses exact, +//! operator-local payload identities for consolidation. //! * **Chunked**: the stash is differential's chunk merge batcher over //! `ColumnChunk`s and the feedback arrangement is a spine of chunk batches. //! Committed chunk bodies spill to the process buffer pool, the drain @@ -77,7 +80,9 @@ //! through the storage-owned column pager, and prior state comes back //! through a trace cursor. //! -//! Both flavors' spill paths are gated by `enable_upsert_paged_spill`. +//! Paged and chunked spill paths are gated by `enable_upsert_paged_spill`. +//! The experimental payload flavor is selected by `enable_upsert_payload_stash` +//! and always allocates its external payloads through the process pool. //! //! ## Eligibility condition (total order) //! @@ -88,6 +93,8 @@ //! with `p < ts` is ineligible (persist hasn't caught up), and one with //! `ts < p` is already persisted and dropped. +mod payload; + use std::fmt::Debug; use differential_dataflow::difference::{IsZero, Semigroup}; @@ -105,7 +112,9 @@ use mz_repr::{Datum, Diff, GlobalId, Row}; #[cfg(feature = "fuzzing")] use mz_row_spine::DatumSeq; use mz_row_spine::{ValRowColPagedBuilder, ValRowSpine}; -use mz_storage_types::dyncfgs::{ENABLE_UPSERT_ASYNC_READS, ENABLE_UPSERT_CHUNKED_STASH}; +use mz_storage_types::dyncfgs::{ + ENABLE_UPSERT_ASYNC_READS, ENABLE_UPSERT_CHUNKED_STASH, ENABLE_UPSERT_PAYLOAD_STASH, +}; use mz_storage_types::errors::{DataflowError, EnvelopeError, UpsertError}; use mz_timely_util::builder_async::{ AsyncOutputHandle, Event as AsyncEvent, OperatorBuilder as AsyncOperatorBuilder, @@ -137,7 +146,7 @@ use crate::upsert::UpsertSourceTime; use crate::upsert::UpsertValue; /// Which stash and feedback-arrangement representation the upsert-v2 -/// operator instantiates. The two flavors run the same operator loop; they +/// operator instantiates. All flavors run the same operator loop. They /// differ in the batcher, the feedback trace, and how the drain reads prior /// state. See the module docs for the comparison. #[derive(Clone, Copy, Debug)] @@ -146,6 +155,8 @@ pub enum UpsertStashFlavor { /// arrangement, cursor-based drain. Spills through the storage-owned /// column pager. Paged, + /// Columnar metadata with independently owned payload blocks. + Payload, /// Chunk merge batcher stash, chunk-spine feedback arrangement, /// bulk-probe drain. Spills through the process buffer pool. Chunked { async_reads: bool }, @@ -156,7 +167,9 @@ impl UpsertStashFlavor { /// source at operator construction time, so a dataflow keeps one flavor /// for its whole life even if the flag flips underneath it. pub fn from_config(config: &ConfigSet) -> Self { - if ENABLE_UPSERT_CHUNKED_STASH.get(config) { + if ENABLE_UPSERT_PAYLOAD_STASH.get(config) { + Self::Payload + } else if ENABLE_UPSERT_CHUNKED_STASH.get(config) { Self::Chunked { async_reads: ENABLE_UPSERT_ASYNC_READS.get(config), } @@ -247,18 +260,18 @@ type FeedbackSpine = ChunkSpine>; // `(key, time)` runs. #[derive(Clone, Debug, Default, columnar::Columnar)] #[columnar(derive(PartialEq, Eq, PartialOrd, Ord))] -struct UpsertDiff { +struct UpsertDiff { from_time: O, - value: Option, + value: Option, } -impl IsZero for UpsertDiff { +impl IsZero for UpsertDiff { fn is_zero(&self) -> bool { false } } -impl Semigroup for UpsertDiff { +impl Semigroup for UpsertDiff { fn plus_equals(&mut self, rhs: &Self) { if rhs.from_time > self.from_time { *self = rhs.clone(); @@ -272,15 +285,16 @@ impl Semigroup for UpsertDiff { // wins" comparison — copying the value `Row` out of the column solely when `rhs` // wins. Losing folds (the common case for a repeatedly-updated key) then pay no // `Row` copy at all. -impl<'a, O> Semigroup>> for UpsertDiff +impl<'a, O, V> Semigroup>> for UpsertDiff where O: columnar::Columnar + Ord + Clone, + V: columnar::Columnar + Clone, { - fn plus_equals(&mut self, rhs: &columnar::Ref<'a, UpsertDiff>) { + fn plus_equals(&mut self, rhs: &columnar::Ref<'a, UpsertDiff>) { let rhs_from_time = ::into_owned(rhs.from_time); if rhs_from_time > self.from_time { self.from_time = rhs_from_time; - self.value = as columnar::Columnar>::into_owned(rhs.value); + self.value = as columnar::Columnar>::into_owned(rhs.value); } } } @@ -288,11 +302,11 @@ where /// One source-stash update: a key, its dataflow time, and the payload diff. /// `O` is the columnar order key projected from the source `FromTime` (see /// [`UpsertSourceTime`]). -type UpsertUpdate = (UpsertKey, T, UpsertDiff); +type UpsertUpdate = (UpsertKey, T, UpsertDiff); /// One stash chunk: a sorted, consolidated run of updates, resident or /// spilled to the buffer pool. -type UpsertChunk = ColumnChunk>; +type UpsertChunk = ColumnChunk>; /// The chunked flavor's stash: differential's chunk merge batcher over /// `ColumnChunk`s. Data is pushed in unsorted. The batcher maintains @@ -307,11 +321,11 @@ type UpsertChunkBatcher = ChunkBatcher>; /// like [`UpsertChunkBatcher`] but storing each chain entry as a `Column` /// routed through the storage-owned pager, which pages cold chains out of /// RSS. -type UpsertPagedBatcher = ColumnMergeBatcher>; +type UpsertPagedBatcher = ColumnMergeBatcher>; /// The chunker that sorts and consolidates raw input into the `Column` chunks /// both stash batchers consume. -type UpsertChunker = ColumnChunker>; +type UpsertChunker = ColumnChunker>; /// The operator's data-output handle. A fueled `Vec` builder so the drain can /// `give_fueled` each emitted update and yield to timely under large snapshot @@ -444,6 +458,35 @@ where source_config.source_statistics.clone(), ); match flavor { + UpsertStashFlavor::Payload => { + let store = payload::store(); + let (encoded, token) = payload::encode_feedback(encoded, store.clone()); + let persist_arranged = arrange_core::< + _, + _, + payload::FeedbackChunker, + ChunkBatcher< + mz_timely_util::columnar::payload::PayloadChunk, + >, + ChunkBuilder< + mz_timely_util::columnar::payload::PayloadChunk, + >, + payload::FeedbackSpine, + >(encoded, Pipeline, "Persist payload feedback"); + let mut persist_token = persist_token.unwrap_or_default(); + persist_token.push(token); + build_upsert_operator::( + input, + resume_upper, + persist_arranged, + Some(persist_token), + upsert_metrics, + source_config, + true, + Some(store), + ) + } + UpsertStashFlavor::Chunked { async_reads } => { // Chains and sealed batches alike are `FeedbackChunk`s whose // bodies spill to the buffer pool, behind the same process spill @@ -464,6 +507,7 @@ where upsert_metrics, source_config, async_reads, + None, ) } UpsertStashFlavor::Paged => { @@ -488,6 +532,7 @@ where upsert_metrics, source_config, false, + None, ) } } @@ -573,6 +618,7 @@ fn build_upsert_operator<'scope, A, T, FromTime>( upsert_metrics: UpsertMetrics, source_config: crate::source::SourceExportCreationConfig, async_reads: bool, + payload_store: Option, ) -> ( VecCollection<'scope, T, Result, Diff>, StreamVec<'scope, T, (Option, HealthStatusUpdate)>, @@ -581,6 +627,7 @@ fn build_upsert_operator<'scope, A, T, FromTime>( ) where A: UpsertStashArm, + for<'a> columnar::Ref<'a, A::Value>: Copy + Ord, T: Timestamp + TotalOrder + Sync, T: Refines + differential_dataflow::lattice::Lattice, T: columnation::Columnation, @@ -634,13 +681,13 @@ where // pushed in, bounding memory to O(unique key-time pairs) even during // large initial snapshots. How cold stash state leaves RSS is // flavor-specific; see the arm impls. - let mut batcher = A::new_batcher(); + let mut batcher = A::new_batcher(payload_store); // The chunker sorts and consolidates raw input into the `Column` chunks // the batcher consumes. - let mut chunker: UpsertChunker = Default::default(); + let mut chunker: UpsertChunker = Default::default(); // Scratch buffer for accumulating source events before flushing to // the batcher. Drained on each iteration via the chunker. - let mut push_buffer: Vec> = Vec::new(); + let mut push_buffer: Vec> = Vec::new(); // Capability held at the minimum time of any buffered data. When // Some, the operator may still produce output; when None, the @@ -684,7 +731,8 @@ where { continue; } - let value = value.as_ref().map(upsert_value_to_row); + let value = + A::encode(value.as_ref().map(upsert_value_to_row), &mut batcher); let from_time = from_time.upsert_order(); push_buffer.push((key, ts, UpsertDiff { from_time, value })); pushed_any = true; @@ -812,6 +860,7 @@ where source_config.worker_id, source_config.id, async_reads, + &mut batcher, ) .await; @@ -884,6 +933,11 @@ where O: columnar::Columnar + Default + Ord + Clone + Send + Sync + 'static, for<'a> columnar::Ref<'a, O>: Ord + Copy, { + type Value: columnar::Columnar + Default + Clone; + + fn encode(value: Option, batcher: &mut Self::Batcher) -> Option; + fn end_flush(_batcher: &mut Self::Batcher) {} + /// The feedback arrangement's spine. `'static` because the operator /// future owns a trace agent for it. type Spine: TraceReader