diff --git a/misc/python/materialize/feature_benchmark/scenarios/mv_sink.py b/misc/python/materialize/feature_benchmark/scenarios/mv_sink.py new file mode 100644 index 0000000000000..be791bae1d3b3 --- /dev/null +++ b/misc/python/materialize/feature_benchmark/scenarios/mv_sink.py @@ -0,0 +1,255 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +from textwrap import dedent + +from materialize.feature_benchmark.action import Action, TdAction +from materialize.feature_benchmark.measurement_source import MeasurementSource, Td +from materialize.feature_benchmark.scenario import Scenario + + +class MvSink(Scenario): + """Feature benchmarks for the materialized view sink's write path. + + These exercise the `write` operator and its correction buffer with the + input shapes that hurt: catching up through many distinct timestamps + after a restart, retraction-heavy steady state, and a large far-future + update mass behind a temporal filter. + + Two dyncfgs select alternative implementations of this path, + `enable_compute_sync_mv_sink` (sync Timely operators feeding Tokio tasks + instead of async operators) and `enable_compute_correction_v2` (the + correction buffer implementation). The nightly run compares against the + merge base with identical settings, so it catches code regressions and a + flag flip, but not a difference between the two settings themselves. To + compare the settings, run the same build against itself with different + parameters: + + bin/mzcompose --find feature-benchmark run default \\ + --root-scenario=MvSink \\ + --other-tag=mzbuild-$(bin/mzimage fingerprint materialized) \\ + --this-params=enable_compute_sync_mv_sink=true \\ + --other-params=enable_compute_sync_mv_sink=false + + Memory matters as much as wall-clock here: the sink buffers every + desired update that persist has not absorbed yet, so a write path that + falls behind shows up as replica memory before it shows up as latency. + Memory is sampled after each measurement, so a backlog that was released + before the trailing `SELECT` returned only shows if the allocator retains + it. + """ + + +class MvSinkCatchUp(MvSink): + """Measure how long the MV sink takes to catch up through many distinct timestamps. + + The view's cluster is taken offline while its input table absorbs many + separate updates, one timestamp each. Bringing the cluster back makes the + sink replay the desired collection from the view's old as-of through every + one of those timestamps while persist writes trail behind, so the + correction buffer holds the whole backlog and is drained one batch at a + time. Work per drained step must stay proportional to that step, otherwise + the catch-up is quadratic in the number of buffered timestamps. + + Runs on a dedicated cluster so its replica can be taken offline. That + replica is a process inside the `materialized` container, so its memory + is accounted to MEMORY_MZ rather than MEMORY_CLUSTERD. + """ + + SCALE = 5 + # Distinct timestamps the sink has to replay when it comes back. + UPDATES = 500 + # Every update touches one row in STRIDE, so each timestamp carries + # 2 * n / STRIDE updates: one retraction and one addition per row. + STRIDE = 100 + + def init(self) -> list[Action]: + return [ + self.table_ten(), + TdAction(dedent(f""" + > CREATE CLUSTER mv_sink_cluster SIZE 'scale=1,workers=1', REPLICATION FACTOR 1 + + > CREATE TABLE t (key INTEGER, v INTEGER) + + > INSERT INTO t SELECT {self.unique_values()}, 0 FROM {self.join()} + """)), + ] + + def before(self) -> Action: + updates = "\n".join( + f"> UPDATE t SET v = v + 1 WHERE key % {self.STRIDE} = 0" + for _ in range(self.UPDATES) + ) + # The view must be created fresh and hydrated before the cluster goes + # offline, so the sink's as-of sits below all of the buffered updates. + return TdAction(dedent(""" + > DROP MATERIALIZED VIEW IF EXISTS mv + + > UPDATE t SET v = 0 + + > ALTER CLUSTER mv_sink_cluster SET (REPLICATION FACTOR 1) + + > CREATE MATERIALIZED VIEW mv IN CLUSTER mv_sink_cluster AS SELECT key, v FROM t + + > SELECT SUM(v) FROM mv + 0 + + > ALTER CLUSTER mv_sink_cluster SET (REPLICATION FACTOR 0) + """) + updates + "\n") + + def benchmark(self) -> MeasurementSource: + return Td(dedent(f""" + > SELECT 1 + /* A */ + 1 + + > ALTER CLUSTER mv_sink_cluster SET (REPLICATION FACTOR 1) + + > SELECT SUM(v) FROM mv + /* B */ + {self.UPDATES * (self.n() // self.STRIDE)} + """)) + + +class MvSinkRetractions(MvSink): + """Measure MV sink latency under retraction-heavy updates to a hydrated view. + + Every `UPDATE` rewrites all rows, so each timestamp carries one addition + and one retraction per row. Each written batch cancels against the next + round of updates, which is where a correction buffer that fails to + consolidate what it has emitted regresses fastest. + + Runs on the default cluster, which the feature-benchmark composition puts + on the external `clusterd` container, so MEMORY_CLUSTERD reflects the + sink's buffering. + """ + + SCALE = 5 + # Full-table rewrites per measurement. + UPDATES = 10 + + def init(self) -> list[Action]: + return [ + self.table_ten(), + TdAction(dedent(f""" + > CREATE TABLE t (key INTEGER, v INTEGER) + + > INSERT INTO t SELECT {self.unique_values()}, 0 FROM {self.join()} + + > CREATE MATERIALIZED VIEW mv AS SELECT key, v FROM t + + > SELECT SUM(v) FROM mv + 0 + """)), + ] + + def before(self) -> Action: + return TdAction(dedent(""" + > UPDATE t SET v = 0 + + > SELECT SUM(v) FROM mv + 0 + """)) + + def benchmark(self) -> MeasurementSource: + updates = "\n".join("> UPDATE t SET v = v + 1" for _ in range(self.UPDATES)) + # Multi-line fragments are appended after dedenting: an indented `>` + # line would read as a continuation of the previous statement. + return Td( + dedent(""" + > SELECT 1 + /* A */ + 1 + + """) + + updates + + dedent(f""" + + > SELECT SUM(v) FROM mv + /* B */ + {self.UPDATES * self.n()} + """) + ) + + +class MvSinkTemporalFilter(MvSink): + """Measure MV sink latency for a temporal filter view with a large far-future update mass. + + Every row's retraction lands at its own far-future time when its window + closes, and deleting a row cancels that retraction at yet another future + time. No batch can ever write those updates, so the correction buffer + holds an ever-growing mass of them. Draining the present, here small + insert-and-delete rounds, must not touch that mass. + + Runs on the default cluster, which the feature-benchmark composition puts + on the external `clusterd` container, so MEMORY_CLUSTERD reflects the + sink's buffering. + """ + + SCALE = 6 + # Insert-and-delete rounds per measurement. + ROUNDS = 10 + # Rows inserted per round. The previous round's rows are deleted alongside. + BATCH = 1000 + + def init(self) -> list[Action]: + return [ + TdAction(dedent(f""" + > CREATE TABLE events (key INTEGER, event_ts TIMESTAMP) + + > INSERT INTO events + SELECT key, TIMESTAMP '2100-01-01' + INTERVAL '1 second' * key + FROM generate_series(1, {self.n()}) AS key + + > CREATE MATERIALIZED VIEW mv AS + SELECT key, event_ts FROM events + WHERE mz_now() <= event_ts + INTERVAL '30 days' + + > SELECT COUNT(*) FROM mv + {self.n()} + """)), + ] + + def before(self) -> Action: + return TdAction(dedent(f""" + > DELETE FROM events WHERE key > {self.n()} + + > SELECT COUNT(*) FROM mv + {self.n()} + """)) + + def benchmark(self) -> MeasurementSource: + rounds = [] + for r in range(self.ROUNDS): + lo = self.n() + r * self.BATCH + 1 + hi = self.n() + (r + 1) * self.BATCH + rounds.append( + f"> INSERT INTO events " + f"SELECT key, TIMESTAMP '2100-01-01' + INTERVAL '1 second' * key " + f"FROM generate_series({lo}, {hi}) AS key" + ) + if r > 0: + rounds.append( + f"> DELETE FROM events WHERE key >= {lo - self.BATCH} AND key < {lo}" + ) + return Td( + dedent(""" + > SELECT 1 + /* A */ + 1 + + """) + + "\n".join(rounds) + + dedent(f""" + + > SELECT COUNT(*) FROM mv + /* B */ + {self.n() + self.BATCH} + """) + ) diff --git a/src/compute/benches/correction.rs b/src/compute/benches/correction.rs index eccc74737fae7..d6ccd4b23cc2b 100644 --- a/src/compute/benches/correction.rs +++ b/src/compute/benches/correction.rs @@ -7,15 +7,11 @@ // the Business Source License, use of this software will be governed // by the Apache License, Version 2.0. -//! Micro-benchmark comparing `CorrectionV1` and `CorrectionV2` on hydration-style -//! workloads, i.e., workloads where the input catches up with the current time by -//! passing through many distinct timestamps. +//! Micro-benchmark comparing `CorrectionV1` and `CorrectionV2` on the hydration-style workloads +//! from `mz_compute::sink::correction_workload`. //! -//! The scenario this models: an MV sink restarts with an old as-of and the desired -//! input replays through `T` distinct timestamps while persist writes (and thus -//! `advance_since`/`updates_before` calls) trail behind. Reads and since advancement -//! must only do work proportional to the drained slice, otherwise the catch-up -//! degenerates into quadratic behavior in `T`. +//! Wall-clock is the measurement here. The deterministic complexity guard for the same workloads +//! lives in the `correction_v2` unit tests, which count structural work instead of time. //! //! Run with: //! @@ -32,10 +28,11 @@ use criterion::{ }; use mz_compute::sink::correction::CorrectionV1; use mz_compute::sink::correction_v2::CorrectionV2; +use mz_compute::sink::correction_workload::{Pattern, make_batches}; use mz_ore::metrics::MetricsRegistry; use mz_persist_client::cfg::PersistConfig; use mz_persist_client::metrics::{Metrics, SinkMetrics}; -use mz_repr::{Datum, Diff, Row, Timestamp}; +use mz_repr::{Diff, Row, Timestamp}; use timely::progress::Antichain; /// Default value of `compute_correction_v2_chain_proportionality`. @@ -45,12 +42,6 @@ const CHUNK_SIZE: usize = 8 * 1024; /// Default value of `consolidating_vec_growth_dampener`. const GROWTH_DAMPENER: usize = 1; -/// Number of updates inserted per distinct timestamp. -const UPDATES_PER_TS: u64 = 16; - -/// Time offset of far-future retractions in the temporal-filter pattern. -const TEMPORAL_OFFSET: u64 = 1 << 40; - #[derive(Clone, Copy)] enum Version { V1, @@ -114,31 +105,6 @@ impl Corr { } } -/// The shape of the update stream fed into the correction buffer. -#[derive(Clone, Copy)] -enum Pattern { - /// Every timestamp appends new, distinct rows. Nothing consolidates away. - Append, - /// Every timestamp updates the same set of keys: an addition for the new value and a - /// retraction of the previous one. Retraction-heavy, consolidates down to a small set. - Upsert, - /// Every timestamp appends new rows accompanied by their far-future retractions, and deletes - /// the previous timestamp's rows, retracting now and re-adding the far-future retraction at a - /// slightly different future time. Models an MV behind a temporal filter (e.g. a last-30-days - /// view): an ever-growing mass of far-future updates that never participates in reads. - TemporalFilter, -} - -impl Pattern { - fn name(self) -> &'static str { - match self { - Self::Append => "append", - Self::Upsert => "upsert", - Self::TemporalFilter => "temporal_filter", - } - } -} - fn sink_metrics() -> SinkMetrics { let registry = MetricsRegistry::new(); let metrics = Metrics::new(&PersistConfig::new_for_tests(), ®istry); @@ -163,59 +129,6 @@ fn make_correction(version: Version, metrics: &SinkMetrics) -> Corr { } } -fn row(key: u64, value: u64) -> Row { - let payload = format!("payload-{value:016}"); - Row::pack_slice(&[Datum::UInt64(key), Datum::String(&payload)]) -} - -/// Generate one batch of updates per distinct timestamp `0..num_ts`. -fn make_batches(num_ts: u64, pattern: Pattern) -> Vec> { - (0..num_ts) - .map(|t| { - let time = Timestamp::from(t); - match pattern { - Pattern::Append => (0..UPDATES_PER_TS) - .map(|i| (row(t * UPDATES_PER_TS + i, t), time, Diff::ONE)) - .collect(), - Pattern::Upsert => (0..UPDATES_PER_TS / 2) - .flat_map(|key| { - let addition = (row(key, t), time, Diff::ONE); - let retraction = t - .checked_sub(1) - .map(|prev| (row(key, prev), time, -Diff::ONE)); - std::iter::once(addition).chain(retraction) - }) - .collect(), - Pattern::TemporalFilter => (0..UPDATES_PER_TS / 4) - .flat_map(|i| { - let key = t * (UPDATES_PER_TS / 4) + i; - // New row, plus its retraction when the temporal filter window closes. - let this = [ - (row(key, t), time, Diff::ONE), - ( - row(key, t), - Timestamp::from(t + TEMPORAL_OFFSET), - -Diff::ONE, - ), - ]; - // Delete the previous timestamp's row: retract it now and cancel its - // window-close retraction. The cancellation lands at a different future - // time than the original retraction, so the far-future mass grows. - let prev = t.checked_sub(1).map(|p| { - let key = p * (UPDATES_PER_TS / 4) + i; - [ - (row(key, p), time, -Diff::ONE), - (row(key, p), Timestamp::from(t + TEMPORAL_OFFSET), Diff::ONE), - ] - }); - this.into_iter().chain(prev.into_iter().flatten()) - }) - .collect(), - } - }) - .collect() -} - /// Fill a fresh correction buffer with all batches, mimicking desired input that has /// run far ahead of persist writes. fn filled_correction( @@ -310,7 +223,7 @@ fn bench_correction(c: &mut Criterion) { let metrics = sink_metrics(); let num_ts_values = [1024, 4096, 16384]; - for pattern in [Pattern::Append, Pattern::Upsert, Pattern::TemporalFilter] { + for pattern in Pattern::ALL { let mut group = c.benchmark_group(format!("correction_drain_stepwise/{}", pattern.name())); configure(&mut group); for num_ts in num_ts_values { diff --git a/src/compute/src/sink.rs b/src/compute/src/sink.rs index 002e19738b05e..48e7bfbd6f69d 100644 --- a/src/compute/src/sink.rs +++ b/src/compute/src/sink.rs @@ -16,6 +16,10 @@ mod correction; pub mod correction_v2; #[cfg(not(feature = "bench"))] mod correction_v2; +#[cfg(feature = "bench")] +pub mod correction_workload; +#[cfg(all(test, not(feature = "bench")))] +mod correction_workload; mod materialized_view; mod materialized_view_v2; mod metric_sink; diff --git a/src/compute/src/sink/correction_v2.rs b/src/compute/src/sink/correction_v2.rs index b8913c38d5112..392d50597473c 100644 --- a/src/compute/src/sink/correction_v2.rs +++ b/src/compute/src/sink/correction_v2.rs @@ -1851,302 +1851,4 @@ impl Ord for MergeCursor { } #[cfg(test)] -mod tests { - use mz_ore::metrics::MetricsRegistry; - use mz_persist_client::cfg::PersistConfig; - use mz_persist_client::metrics::Metrics; - use mz_repr::{Diff, Timestamp}; - - use super::*; - use crate::sink::correction::CorrectionV1; - - #[mz_ore::test] - fn chain_builder_update_count_matches_items() { - let mut builder = ChainBuilder::::default(); - for i in 0..10_u64 { - let d = i64::try_from(i).expect("fits"); - builder.push_owned(&(d, Timestamp::new(i), Diff::ONE)); - } - let chain = builder.finish(); - assert_eq!(chain.update_count, chain.iter().count()); - } - - /// Push enough updates to cross at least one `mint()` boundary, forcing the - /// `Align` encode -> `from_bytes` decode roundtrip (the spilling path this data - /// structure exists to support), and assert `iter()` roundtrips values, order, - /// and diffs across the spill boundary. - #[mz_ore::test] - #[cfg_attr(miri, ignore)] // too slow: crossing the ~2 MiB mint boundary needs ~200k updates - fn chain_builder_roundtrips_across_mint_boundary() { - // A single `mint()` fires near the ~2 MiB (`SHIP_WORDS`) serialized boundary. With - // three 8-byte columns per update that's tens of thousands of updates; pushing 200k - // comfortably forces multiple mints. - let count = 200_000_u64; - - let mut builder = ChainBuilder::::default(); - for i in 0..count { - let d = i64::try_from(i).expect("fits"); - builder.push_owned(&(d, Timestamp::new(i), Diff::ONE)); - } - let chain = builder.finish(); - - // Crossing the mint boundary must have produced more than one chunk; otherwise the spill - // path (each minted chunk is paged out and read back through the pager) wouldn't be - // exercised. The chunk payload itself is now behind the pager (see [`Chunk`]), so we - // assert on chunk count rather than inspecting the column variant directly. - assert!( - chain.chunks.len() > 1, - "expected multiple minted chunks, got {} chunk(s): {:?}", - chain.chunks.len(), - chain.chunks, - ); - - // `iter()` must roundtrip every update, in order, with correct diffs. - assert_eq!(chain.update_count, usize::try_from(count).expect("fits")); - let mut expected = 0_u64; - for (d, t, r) in chain.iter() { - assert_eq!(d, i64::try_from(expected).expect("fits")); - assert_eq!(t, Timestamp::new(expected)); - assert_eq!(r, Diff::ONE); - expected += 1; - } - assert_eq!(expected, count); - } - - fn sink_metrics() -> SinkMetrics { - let registry = MetricsRegistry::new(); - let metrics = Metrics::new(&PersistConfig::new_for_tests(), ®istry); - metrics.sink.clone() - } - - /// Run the same stepwise-drain workload through `CorrectionV1` and `CorrectionV2` and assert - /// that they emit the same updates at every step. - /// - /// Models the `write_batches` operator catching up through many distinct timestamps: the - /// desired input runs ahead, batches are written one timestamp at a time, and written updates - /// come back negated through the persist feedback. - #[mz_ore::test] - // Columnation regions are not Stacked Borrows compliant: later pushes invalidate the - // provenance of previously stored items under Miri. - #[cfg_attr(miri, ignore)] - fn equivalence_with_v1() { - let sink_metrics = sink_metrics(); - - let mut v1 = - CorrectionV1::::new(sink_metrics.clone(), sink_metrics.for_worker(0), 1); - let mut v2 = CorrectionV2::::new( - sink_metrics.clone(), - sink_metrics.for_worker(0), - None, - 3.0, - 8 * 1024, - ); - - let num_ts = 50; - let keys = 4; - - // Upsert-style input: every timestamp updates each key, retracting the previous value. - let batch = |t: u64| -> Vec<(String, Timestamp, Diff)> { - (0..keys) - .flat_map(|k| { - let addition = (format!("{k}-{t}"), Timestamp::from(t), Diff::ONE); - let retraction = t - .checked_sub(1) - .map(|p| (format!("{k}-{p}"), Timestamp::from(t), -Diff::ONE)); - std::iter::once(addition).chain(retraction) - }) - .collect() - }; - - // Pre-fill both with all batches, like a catch-up where the input runs ahead. - for t in 0..num_ts { - v1.insert(&mut batch(t)); - v2.insert(&mut batch(t)); - } - - // Drain stepwise, with persist feedback, comparing emissions. - for t in 0..num_ts { - let upper = Antichain::from_elem(Timestamp::from(t + 1)); - - let mut out1: Vec<_> = v1.updates_before(&upper).collect(); - let mut out2: Vec<_> = v2.updates_before(&upper).collect(); - out1.sort(); - out2.sort(); - assert_eq!(out1, out2, "diverged at t={t}"); - - v1.insert_negated(&mut out1.clone()); - v2.insert_negated(&mut out2); - v1.advance_since(upper.clone()); - v2.advance_since(upper); - } - - // Compare the final state at the since. - let upper = Antichain::from_elem(Timestamp::from(num_ts + 1)); - v1.consolidate_at_since(); - v2.consolidate_at_since(); - let mut out1: Vec<_> = v1.updates_before(&upper).collect(); - let mut out2: Vec<_> = v2.updates_before(&upper).collect(); - out1.sort(); - out2.sort(); - assert_eq!(out1, out2); - } - - /// A since jump across many distinct buffered timestamps must collapse them onto the since. - #[mz_ore::test] - // Columnation regions are not Stacked Borrows compliant: later pushes invalidate the - // provenance of previously stored items under Miri. - #[cfg_attr(miri, ignore)] - fn since_jump() { - let sink_metrics = sink_metrics(); - let mut v2 = CorrectionV2::::new( - sink_metrics.clone(), - sink_metrics.for_worker(0), - None, - 3.0, - 8 * 1024, - ); - - let num_ts = 100; - for t in 0..num_ts { - v2.insert(&mut vec![ - (format!("a-{t}"), Timestamp::from(t), Diff::ONE), - (format!("a-{t}"), Timestamp::from(t), -Diff::ONE), - (format!("b-{t}"), Timestamp::from(t), Diff::ONE), - ]); - } - - v2.advance_since(Antichain::from_elem(Timestamp::from(num_ts))); - v2.consolidate_at_since(); - - let upper = Antichain::from_elem(Timestamp::from(num_ts + 1)); - let out: Vec<_> = v2.updates_before(&upper).collect(); - assert_eq!(out.len(), usize::try_from(num_ts).unwrap()); - assert!( - out.iter() - .all(|(_, t, r)| *t == Timestamp::from(num_ts) && *r == Diff::ONE) - ); - } - - /// Reads must not observe updates at or beyond their `upper`, even when the `upper` is not - /// beyond the `since`. - #[mz_ore::test] - // Columnation regions are not Stacked Borrows compliant: later pushes invalidate the - // provenance of previously stored items under Miri. - #[cfg_attr(miri, ignore)] - fn upper_not_beyond_since() { - let sink_metrics = sink_metrics(); - let mut v2 = CorrectionV2::::new( - sink_metrics.clone(), - sink_metrics.for_worker(0), - None, - 3.0, - 8 * 1024, - ); - - v2.insert(&mut vec![( - "a".to_owned(), - Timestamp::from(5_u64), - Diff::ONE, - )]); - v2.advance_since(Antichain::from_elem(Timestamp::from(10_u64))); - - // The update logically lives at time 10 now, so a read before 7 must be empty. - let upper = Antichain::from_elem(Timestamp::from(7_u64)); - assert_eq!(v2.updates_before(&upper).count(), 0); - - // A read before 11 must emit it, advanced to the since. - let upper = Antichain::from_elem(Timestamp::from(11_u64)); - let out: Vec<_> = v2.updates_before(&upper).collect(); - assert_eq!( - out, - vec![("a".to_owned(), Timestamp::from(10_u64), Diff::ONE)] - ); - } - - /// A [`PagingPolicy`] that always spills to the swap backend, uncompressed. - /// - /// The default global pager keeps every chunk resident; installing this drives the actual - /// spill path so the tests exercise [`Chunk::column`]'s page-in through [`mz_ore::pager`]. - /// - /// [`PagingPolicy`]: column_pager::PagingPolicy - struct ForceSwap; - - impl column_pager::PagingPolicy for ForceSwap { - fn decide(&self, _hint: column_pager::PageHint) -> column_pager::PageDecision { - column_pager::PageDecision::Page { - backend: mz_ore::pager::Backend::Swap, - codec: None, - } - } - fn record(&self, _event: column_pager::PageEvent) {} - } - - /// Install a global pager that spills every chunk to swap for the duration of `f`, then - /// restore the default (disabled) pager. The global pager is process-wide; concurrent tests - /// only ever observe a correct round-trip regardless of backend, so racing on it is benign. - fn with_swap_pager(f: impl FnOnce() -> R) -> R { - use std::sync::Arc; - column_pager::set_global_pager(column_pager::ColumnPager::new(Arc::new(ForceSwap))); - let result = f(); - column_pager::set_global_pager(column_pager::ColumnPager::disabled()); - result - } - - /// Build a chain crossing the mint boundary while every chunk is spilled to swap, then assert - /// `iter()` (the read path behind `updates_before`) pages each chunk back in and roundtrips - /// values, order, and diffs. - #[mz_ore::test] - #[cfg_attr(miri, ignore)] // madvise on the swap backend is unsupported under miri - fn iter_roundtrips_through_swap_backend() { - let count = 200_000_u64; - with_swap_pager(|| { - let mut builder = ChainBuilder::::default(); - for i in 0..count { - let d = i64::try_from(i).expect("fits"); - builder.push_owned(&(d, Timestamp::new(i), Diff::ONE)); - } - let chain = builder.finish(); - assert!(chain.chunks.len() > 1, "expected multiple minted chunks"); - assert_eq!(chain.update_count, usize::try_from(count).expect("fits")); - - let mut expected = 0_u64; - for (d, t, r) in chain.iter() { - assert_eq!(d, i64::try_from(expected).expect("fits")); - assert_eq!(t, Timestamp::new(expected)); - assert_eq!(r, Diff::ONE); - expected += 1; - } - assert_eq!(expected, count); - }); - } - - /// Drive a [`Cursor`] over a spilled, multi-chunk chain to completion (the access pattern - /// merges use). Each step pages the front chunk back in via [`Chunk::column`]; assert the - /// cursor yields every update in order. - #[mz_ore::test] - #[cfg_attr(miri, ignore)] // madvise on the swap backend is unsupported under miri - fn cursor_steps_through_swap_backend() { - let count = 200_000_u64; - with_swap_pager(|| { - let mut builder = ChainBuilder::::default(); - for i in 0..count { - let d = i64::try_from(i).expect("fits"); - builder.push_owned(&(d, Timestamp::new(i), Diff::ONE)); - } - let chain = builder.finish(); - assert!(chain.chunks.len() > 1, "expected multiple minted chunks"); - - let mut rest = chain.into_cursor(); - let mut expected = 0_u64; - while let Some(cursor) = rest.take() { - let (d, t, r) = cursor.get(); - assert_eq!(i64::into_owned(d), i64::try_from(expected).expect("fits")); - assert_eq!(t, Timestamp::new(expected)); - assert_eq!(r, Diff::ONE); - expected += 1; - rest = cursor.step(); - } - assert_eq!(expected, count); - }); - } -} +mod tests; diff --git a/src/compute/src/sink/correction_v2/tests.rs b/src/compute/src/sink/correction_v2/tests.rs new file mode 100644 index 0000000000000..1d1d1e57b2a24 --- /dev/null +++ b/src/compute/src/sink/correction_v2/tests.rs @@ -0,0 +1,416 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +use mz_ore::metrics::MetricsRegistry; +use mz_persist_client::cfg::PersistConfig; +use mz_persist_client::metrics::Metrics; +use mz_repr::{Diff, Row, Timestamp}; + +use super::*; +use crate::sink::correction::{CorrectionV1, LoggingEvent}; +use crate::sink::correction_workload::{Pattern, make_batches}; + +#[mz_ore::test] +fn chain_builder_update_count_matches_items() { + let mut builder = ChainBuilder::::default(); + for i in 0..10_u64 { + let d = i64::try_from(i).expect("fits"); + builder.push_owned(&(d, Timestamp::new(i), Diff::ONE)); + } + let chain = builder.finish(); + assert_eq!(chain.update_count, chain.iter().count()); +} + +/// Push enough updates to cross at least one `mint()` boundary, forcing the +/// `Align` encode -> `from_bytes` decode roundtrip (the spilling path this data +/// structure exists to support), and assert `iter()` roundtrips values, order, +/// and diffs across the spill boundary. +#[mz_ore::test] +#[cfg_attr(miri, ignore)] // too slow: crossing the ~2 MiB mint boundary needs ~200k updates +fn chain_builder_roundtrips_across_mint_boundary() { + // A single `mint()` fires near the ~2 MiB (`SHIP_WORDS`) serialized boundary. With + // three 8-byte columns per update that's tens of thousands of updates; pushing 200k + // comfortably forces multiple mints. + let count = 200_000_u64; + + let mut builder = ChainBuilder::::default(); + for i in 0..count { + let d = i64::try_from(i).expect("fits"); + builder.push_owned(&(d, Timestamp::new(i), Diff::ONE)); + } + let chain = builder.finish(); + + // Crossing the mint boundary must have produced more than one chunk; otherwise the spill + // path (each minted chunk is paged out and read back through the pager) wouldn't be + // exercised. The chunk payload itself is now behind the pager (see [`Chunk`]), so we + // assert on chunk count rather than inspecting the column variant directly. + assert!( + chain.chunks.len() > 1, + "expected multiple minted chunks, got {} chunk(s): {:?}", + chain.chunks.len(), + chain.chunks, + ); + + // `iter()` must roundtrip every update, in order, with correct diffs. + assert_eq!(chain.update_count, usize::try_from(count).expect("fits")); + let mut expected = 0_u64; + for (d, t, r) in chain.iter() { + assert_eq!(d, i64::try_from(expected).expect("fits")); + assert_eq!(t, Timestamp::new(expected)); + assert_eq!(r, Diff::ONE); + expected += 1; + } + assert_eq!(expected, count); +} + +fn sink_metrics() -> SinkMetrics { + let registry = MetricsRegistry::new(); + let metrics = Metrics::new(&PersistConfig::new_for_tests(), ®istry); + metrics.sink.clone() +} + +/// Run the same stepwise-drain workload through `CorrectionV1` and `CorrectionV2` and assert +/// that they emit the same updates at every step. +/// +/// Models the `write_batches` operator catching up through many distinct timestamps: the +/// desired input runs ahead, batches are written one timestamp at a time, and written updates +/// come back negated through the persist feedback. +#[mz_ore::test] +// Columnation regions are not Stacked Borrows compliant: later pushes invalidate the +// provenance of previously stored items under Miri. +#[cfg_attr(miri, ignore)] +fn equivalence_with_v1() { + let sink_metrics = sink_metrics(); + + let mut v1 = CorrectionV1::::new(sink_metrics.clone(), sink_metrics.for_worker(0), 1); + let mut v2 = CorrectionV2::::new( + sink_metrics.clone(), + sink_metrics.for_worker(0), + None, + 3.0, + 8 * 1024, + ); + + let num_ts = 50; + let keys = 4; + + // Upsert-style input: every timestamp updates each key, retracting the previous value. + let batch = |t: u64| -> Vec<(String, Timestamp, Diff)> { + (0..keys) + .flat_map(|k| { + let addition = (format!("{k}-{t}"), Timestamp::from(t), Diff::ONE); + let retraction = t + .checked_sub(1) + .map(|p| (format!("{k}-{p}"), Timestamp::from(t), -Diff::ONE)); + std::iter::once(addition).chain(retraction) + }) + .collect() + }; + + // Pre-fill both with all batches, like a catch-up where the input runs ahead. + for t in 0..num_ts { + v1.insert(&mut batch(t)); + v2.insert(&mut batch(t)); + } + + // Drain stepwise, with persist feedback, comparing emissions. + for t in 0..num_ts { + let upper = Antichain::from_elem(Timestamp::from(t + 1)); + + let mut out1: Vec<_> = v1.updates_before(&upper).collect(); + let mut out2: Vec<_> = v2.updates_before(&upper).collect(); + out1.sort(); + out2.sort(); + assert_eq!(out1, out2, "diverged at t={t}"); + + v1.insert_negated(&mut out1.clone()); + v2.insert_negated(&mut out2); + v1.advance_since(upper.clone()); + v2.advance_since(upper); + } + + // Compare the final state at the since. + let upper = Antichain::from_elem(Timestamp::from(num_ts + 1)); + v1.consolidate_at_since(); + v2.consolidate_at_since(); + let mut out1: Vec<_> = v1.updates_before(&upper).collect(); + let mut out2: Vec<_> = v2.updates_before(&upper).collect(); + out1.sort(); + out2.sort(); + assert_eq!(out1, out2); +} + +/// A since jump across many distinct buffered timestamps must collapse them onto the since. +#[mz_ore::test] +// Columnation regions are not Stacked Borrows compliant: later pushes invalidate the +// provenance of previously stored items under Miri. +#[cfg_attr(miri, ignore)] +fn since_jump() { + let sink_metrics = sink_metrics(); + let mut v2 = CorrectionV2::::new( + sink_metrics.clone(), + sink_metrics.for_worker(0), + None, + 3.0, + 8 * 1024, + ); + + let num_ts = 100; + for t in 0..num_ts { + v2.insert(&mut vec![ + (format!("a-{t}"), Timestamp::from(t), Diff::ONE), + (format!("a-{t}"), Timestamp::from(t), -Diff::ONE), + (format!("b-{t}"), Timestamp::from(t), Diff::ONE), + ]); + } + + v2.advance_since(Antichain::from_elem(Timestamp::from(num_ts))); + v2.consolidate_at_since(); + + let upper = Antichain::from_elem(Timestamp::from(num_ts + 1)); + let out: Vec<_> = v2.updates_before(&upper).collect(); + assert_eq!(out.len(), usize::try_from(num_ts).unwrap()); + assert!( + out.iter() + .all(|(_, t, r)| *t == Timestamp::from(num_ts) && *r == Diff::ONE) + ); +} + +/// Reads must not observe updates at or beyond their `upper`, even when the `upper` is not +/// beyond the `since`. +#[mz_ore::test] +// Columnation regions are not Stacked Borrows compliant: later pushes invalidate the +// provenance of previously stored items under Miri. +#[cfg_attr(miri, ignore)] +fn upper_not_beyond_since() { + let sink_metrics = sink_metrics(); + let mut v2 = CorrectionV2::::new( + sink_metrics.clone(), + sink_metrics.for_worker(0), + None, + 3.0, + 8 * 1024, + ); + + v2.insert(&mut vec![( + "a".to_owned(), + Timestamp::from(5_u64), + Diff::ONE, + )]); + v2.advance_since(Antichain::from_elem(Timestamp::from(10_u64))); + + // The update logically lives at time 10 now, so a read before 7 must be empty. + let upper = Antichain::from_elem(Timestamp::from(7_u64)); + assert_eq!(v2.updates_before(&upper).count(), 0); + + // A read before 11 must emit it, advanced to the since. + let upper = Antichain::from_elem(Timestamp::from(11_u64)); + let out: Vec<_> = v2.updates_before(&upper).collect(); + assert_eq!( + out, + vec![("a".to_owned(), Timestamp::from(10_u64), Diff::ONE)] + ); +} + +fn default_v2(logging: Option) -> CorrectionV2 { + let sink_metrics = sink_metrics(); + CorrectionV2::new( + sink_metrics.clone(), + sink_metrics.for_worker(0), + logging, + 3.0, + 8 * 1024, + ) +} + +/// Structural work performed while running `f`: the number of updates copied into newly built +/// chains, as reported through the introspection logging hook. +/// +/// Every merge, split, stage flush, and emitted chain reports the chain it produces, so this +/// tracks the cost of maintaining the buffer without depending on wall-clock time. Cursor +/// stepping and time advancement are not reported, but they only ever run over chains that +/// are subsequently rebuilt and thus counted here. +fn chain_work(f: impl FnOnce(ChannelLogging) -> CorrectionV2) -> usize { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let correction = f(ChannelLogging::new(tx)); + drop(correction); + + let mut created = 0; + while let Ok(event) = rx.try_recv() { + if let LoggingEvent::ChainCreated(len) = event { + created += len; + } + } + created +} + +/// Fill a buffer with `num_ts` timestamps of `pattern`, mimicking desired input that ran far +/// ahead of persist, then drain it one timestamp at a time with persist feedback, like the +/// `write_batches` operator does while persist writes catch up step by step. +fn drain_stepwise_work(pattern: Pattern, num_ts: u64) -> usize { + let batches = make_batches(num_ts, pattern); + chain_work(|logging| { + let mut correction = default_v2::(Some(logging)); + for mut batch in batches { + correction.insert(&mut batch); + } + for t in 0..num_ts { + let upper = Antichain::from_elem(Timestamp::from(t + 1)); + // Written updates come back negated through the persist input. Without this + // feedback the buffer would legitimately re-emit everything on every step. + let mut written: Vec<_> = correction.updates_before(&upper).collect(); + correction.insert_negated(&mut written); + correction.advance_since(upper); + } + correction + }) +} + +/// Fill a buffer like [`drain_stepwise_work`], then jump the since across all buffered times +/// at once and read everything, like a sink whose persist shard is already far ahead of the +/// dataflow's as-of. +fn advance_jump_work(pattern: Pattern, num_ts: u64) -> usize { + let batches = make_batches(num_ts, pattern); + chain_work(|logging| { + let mut correction = default_v2::(Some(logging)); + for mut batch in batches { + correction.insert(&mut batch); + } + correction.advance_since(Antichain::from_elem(Timestamp::from(num_ts))); + correction.consolidate_at_since(); + let upper = Antichain::from_elem(Timestamp::from(num_ts + 1)); + let _ = correction.updates_before(&upper).count(); + correction + }) +} + +/// Catching up through `T` distinct timestamps must cost work linear in `T`. +/// +/// The failure mode is a sink restarting with an old as-of where every drained step re-touches +/// everything buffered so far, which turns the catch-up quadratic. Compare structural work +/// between `T` and `4T` timestamps: linear scaling gives a ratio near 4 (chain merges add a +/// logarithmic factor on top; all routines measure 4.0 to 4.2 at these sizes), quadratic +/// scaling gives 16. The temporal-filter pattern additionally checks that the growing +/// far-future mass stays out of the drained slices. +#[mz_ore::test] +// Columnation regions are not Stacked Borrows compliant: later pushes invalidate the +// provenance of previously stored items under Miri. +#[cfg_attr(miri, ignore)] +fn catch_up_work_is_linear() { + const SMALL: u64 = 512; + const LARGE: u64 = 4 * SMALL; + const MAX_RATIO: f64 = 8.0; + + let routines: [(&str, fn(Pattern, u64) -> usize); 2] = [ + ("drain_stepwise", drain_stepwise_work), + ("advance_jump", advance_jump_work), + ]; + + for pattern in Pattern::ALL { + for (routine, work) in routines { + let small = work(pattern, SMALL); + let large = work(pattern, LARGE); + assert!(small > 0, "{routine}/{}: no work recorded", pattern.name()); + let ratio = f64::cast_lossy(large) / f64::cast_lossy(small); + assert!( + ratio <= MAX_RATIO, + "{routine}/{}: work grew {ratio:.1}x for 4x more timestamps \ + ({small} -> {large} updates copied), catch-up is no longer linear", + pattern.name(), + ); + } + } +} + +/// A [`PagingPolicy`] that always spills to the swap backend, uncompressed. +/// +/// The default global pager keeps every chunk resident; installing this drives the actual +/// spill path so the tests exercise [`Chunk::column`]'s page-in through [`mz_ore::pager`]. +/// +/// [`PagingPolicy`]: column_pager::PagingPolicy +struct ForceSwap; + +impl column_pager::PagingPolicy for ForceSwap { + fn decide(&self, _hint: column_pager::PageHint) -> column_pager::PageDecision { + column_pager::PageDecision::Page { + backend: mz_ore::pager::Backend::Swap, + codec: None, + } + } + fn record(&self, _event: column_pager::PageEvent) {} +} + +/// Install a global pager that spills every chunk to swap for the duration of `f`, then +/// restore the default (disabled) pager. The global pager is process-wide; concurrent tests +/// only ever observe a correct round-trip regardless of backend, so racing on it is benign. +fn with_swap_pager(f: impl FnOnce() -> R) -> R { + use std::sync::Arc; + column_pager::set_global_pager(column_pager::ColumnPager::new(Arc::new(ForceSwap))); + let result = f(); + column_pager::set_global_pager(column_pager::ColumnPager::disabled()); + result +} + +/// Build a chain crossing the mint boundary while every chunk is spilled to swap, then assert +/// `iter()` (the read path behind `updates_before`) pages each chunk back in and roundtrips +/// values, order, and diffs. +#[mz_ore::test] +#[cfg_attr(miri, ignore)] // madvise on the swap backend is unsupported under miri +fn iter_roundtrips_through_swap_backend() { + let count = 200_000_u64; + with_swap_pager(|| { + let mut builder = ChainBuilder::::default(); + for i in 0..count { + let d = i64::try_from(i).expect("fits"); + builder.push_owned(&(d, Timestamp::new(i), Diff::ONE)); + } + let chain = builder.finish(); + assert!(chain.chunks.len() > 1, "expected multiple minted chunks"); + assert_eq!(chain.update_count, usize::try_from(count).expect("fits")); + + let mut expected = 0_u64; + for (d, t, r) in chain.iter() { + assert_eq!(d, i64::try_from(expected).expect("fits")); + assert_eq!(t, Timestamp::new(expected)); + assert_eq!(r, Diff::ONE); + expected += 1; + } + assert_eq!(expected, count); + }); +} + +/// Drive a [`Cursor`] over a spilled, multi-chunk chain to completion (the access pattern +/// merges use). Each step pages the front chunk back in via [`Chunk::column`]; assert the +/// cursor yields every update in order. +#[mz_ore::test] +#[cfg_attr(miri, ignore)] // madvise on the swap backend is unsupported under miri +fn cursor_steps_through_swap_backend() { + let count = 200_000_u64; + with_swap_pager(|| { + let mut builder = ChainBuilder::::default(); + for i in 0..count { + let d = i64::try_from(i).expect("fits"); + builder.push_owned(&(d, Timestamp::new(i), Diff::ONE)); + } + let chain = builder.finish(); + assert!(chain.chunks.len() > 1, "expected multiple minted chunks"); + + let mut rest = chain.into_cursor(); + let mut expected = 0_u64; + while let Some(cursor) = rest.take() { + let (d, t, r) = cursor.get(); + assert_eq!(i64::into_owned(d), i64::try_from(expected).expect("fits")); + assert_eq!(t, Timestamp::new(expected)); + assert_eq!(r, Diff::ONE); + expected += 1; + rest = cursor.step(); + } + assert_eq!(expected, count); + }); +} diff --git a/src/compute/src/sink/correction_workload.rs b/src/compute/src/sink/correction_workload.rs new file mode 100644 index 0000000000000..85bb4816e0ac0 --- /dev/null +++ b/src/compute/src/sink/correction_workload.rs @@ -0,0 +1,110 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +//! Hydration-style update streams for exercising the MV sink correction buffers. +//! +//! The scenario these model: an MV sink restarts with an old as-of and the desired input replays +//! through `T` distinct timestamps while persist writes (and thus `advance_since` and +//! `updates_before` calls) trail behind. Reads and since advancement must only do work +//! proportional to the drained slice, otherwise the catch-up degenerates into quadratic behavior +//! in `T`. +//! +//! Shared between the `correction` criterion bench, which compares wall-clock time of the buffer +//! implementations, and the deterministic complexity tests in `correction_v2`, so both measure +//! the same workloads. + +use mz_repr::{Datum, Diff, Row, Timestamp}; + +/// Number of updates inserted per distinct timestamp. +pub const UPDATES_PER_TS: u64 = 16; + +/// Time offset of far-future retractions in the temporal-filter pattern. +const TEMPORAL_OFFSET: u64 = 1 << 40; + +/// The shape of the update stream fed into the correction buffer. +#[derive(Clone, Copy, Debug)] +pub enum Pattern { + /// Every timestamp appends new, distinct rows. Nothing consolidates away. + Append, + /// Every timestamp updates the same set of keys: an addition for the new value and a + /// retraction of the previous one. Retraction-heavy, consolidates down to a small set. + Upsert, + /// Every timestamp appends new rows accompanied by their far-future retractions, and deletes + /// the previous timestamp's rows, retracting now and re-adding the far-future retraction at a + /// slightly different future time. Models an MV behind a temporal filter (e.g. a last-30-days + /// view): an ever-growing mass of far-future updates that never participates in reads. + TemporalFilter, +} + +impl Pattern { + /// All patterns, for iterating over the workloads. + pub const ALL: [Pattern; 3] = [Self::Append, Self::Upsert, Self::TemporalFilter]; + + /// Name for use in benchmark IDs and assertion messages. + pub fn name(self) -> &'static str { + match self { + Self::Append => "append", + Self::Upsert => "upsert", + Self::TemporalFilter => "temporal_filter", + } + } +} + +fn row(key: u64, value: u64) -> Row { + let payload = format!("payload-{value:016}"); + Row::pack_slice(&[Datum::UInt64(key), Datum::String(&payload)]) +} + +/// Generate one batch of updates per distinct timestamp `0..num_ts`. +pub fn make_batches(num_ts: u64, pattern: Pattern) -> Vec> { + (0..num_ts) + .map(|t| { + let time = Timestamp::from(t); + match pattern { + Pattern::Append => (0..UPDATES_PER_TS) + .map(|i| (row(t * UPDATES_PER_TS + i, t), time, Diff::ONE)) + .collect(), + Pattern::Upsert => (0..UPDATES_PER_TS / 2) + .flat_map(|key| { + let addition = (row(key, t), time, Diff::ONE); + let retraction = t + .checked_sub(1) + .map(|prev| (row(key, prev), time, -Diff::ONE)); + std::iter::once(addition).chain(retraction) + }) + .collect(), + Pattern::TemporalFilter => (0..UPDATES_PER_TS / 4) + .flat_map(|i| { + let key = t * (UPDATES_PER_TS / 4) + i; + // New row, plus its retraction when the temporal filter window closes. + let this = [ + (row(key, t), time, Diff::ONE), + ( + row(key, t), + Timestamp::from(t + TEMPORAL_OFFSET), + -Diff::ONE, + ), + ]; + // Delete the previous timestamp's row: retract it now and cancel its + // window-close retraction. The cancellation lands at a different future + // time than the original retraction, so the far-future mass grows. + let prev = t.checked_sub(1).map(|p| { + let key = p * (UPDATES_PER_TS / 4) + i; + [ + (row(key, p), time, -Diff::ONE), + (row(key, p), Timestamp::from(t + TEMPORAL_OFFSET), Diff::ONE), + ] + }); + this.into_iter().chain(prev.into_iter().flatten()) + }) + .collect(), + } + }) + .collect() +} diff --git a/test/feature-benchmark/mzcompose.py b/test/feature-benchmark/mzcompose.py index 88fd9d8767477..e8c90f4cc544c 100644 --- a/test/feature-benchmark/mzcompose.py +++ b/test/feature-benchmark/mzcompose.py @@ -78,6 +78,7 @@ ) from materialize.feature_benchmark.scenarios.concurrency import * # noqa: F401 F403 from materialize.feature_benchmark.scenarios.customer import * # noqa: F401 F403 +from materialize.feature_benchmark.scenarios.mv_sink import * # noqa: F401 F403 from materialize.feature_benchmark.scenarios.optbench import * # noqa: F401 F403 from materialize.feature_benchmark.scenarios.scale import * # noqa: F401 F403 from materialize.feature_benchmark.scenarios.skew import * # noqa: F401 F403