Skip to content

compute: remove the correction v1 buffer #38484

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
antiguru wants to merge 2 commits into
base: main
Choose a base branch
from
Open

compute: remove the correction v1 buffer #38484

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
210aff9
compute: remove the correction v1 buffer
antiguru Aug 26, 2026
937807e
compute: unlink private CorrectionLogger from module docs
antiguru Aug 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions misc/python/materialize/mzcompose/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,6 @@ def get_minimal_system_parameters(
"enable_cast_elimination": "true",
"enable_coalesce_case_transform": "true",
"enable_columnation_lgalloc": "false",
"enable_compute_correction_v2": "true",
"enable_compute_logical_backpressure": "true",
"enable_connection_validation_syntax": "true",
"enable_create_table_from_source": "true",
Expand Down Expand Up @@ -643,7 +642,6 @@ def get_default_system_parameters(
"compute_dataflow_max_inflight_bytes_cc",
"compute_flat_map_fuel",
"compute_temporal_bucketing_summary",
"consolidating_vec_growth_dampener",
"copy_to_s3_parquet_row_group_file_ratio",
"copy_to_s3_arrow_builder_buffer_ratio",
"copy_to_s3_multipart_part_size_bytes",
Expand Down
2 changes: 0 additions & 2 deletions misc/python/materialize/parallel_workload/action.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3187,7 +3187,6 @@ def __init__(
"persist_blob_hedged_get_warm_interval",
"enable_compute_half_join2",
"enable_mz_join_core",
"enable_compute_correction_v2",
"linear_join_yielding",
"enable_lgalloc",
"enable_lgalloc_eager_reclamation",
Expand All @@ -3207,7 +3206,6 @@ def __init__(
"compute_dataflow_max_inflight_bytes_cc",
"subscribe_max_buffered_bytes",
"compute_flat_map_fuel",
"consolidating_vec_growth_dampener",
"compute_hydration_concurrency",
"copy_to_s3_parquet_row_group_file_ratio",
"copy_to_s3_arrow_builder_buffer_ratio",
Expand Down
19 changes: 0 additions & 19 deletions src/compute-types/src/dyncfgs.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,14 +262,6 @@ pub const ENABLE_SYNC_MV_SINK: Config<bool> = Config::new(
ParameterScope::Environment,
);

/// Whether rendering should use the new MV sink correction buffer implementation.
pub const ENABLE_CORRECTION_V2: Config<bool> = Config::new(
"enable_compute_correction_v2",
true,
"Whether compute should use the new MV sink correction buffer implementation.",
ParameterScope::Environment,
);

/// The size factor of subsequent chains in the correction V2 buffer.
pub const CORRECTION_V2_CHAIN_PROPORTIONALITY: Config<f64> = Config::new(
"compute_correction_v2_chain_proportionality",
Expand Down Expand Up @@ -422,15 +414,6 @@ pub const DATAFLOW_MAX_INFLIGHT_BYTES_CC: Config<Option<usize>> = Config::new(
ParameterScope::Replica,
);

/// The term `n` in the growth rate `1 + 1/(n + 1)` for `ConsolidatingVec`.
/// The smallest value `0` corresponds to the greatest allowed growth, of doubling.
pub const CONSOLIDATING_VEC_GROWTH_DAMPENER: Config<usize> = Config::new(
"consolidating_vec_growth_dampener",
1,
"Dampener in growth rate for consolidating vector size",
ParameterScope::Replica,
);

/// The number of dataflows that may hydrate concurrently.
///
/// Enforced in `environmentd`, by the controller's per-replica hydration
Expand Down Expand Up @@ -688,7 +671,6 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
.add(&ENABLE_ERROR_DISTINCT)
.add(&ENABLE_MZ_JOIN_CORE)
.add(&ENABLE_SYNC_MV_SINK)
.add(&ENABLE_CORRECTION_V2)
.add(&CORRECTION_V2_CHAIN_PROPORTIONALITY)
.add(&CORRECTION_V2_CHUNK_SIZE)
.add(&ENABLE_COMPUTE_TEMPORAL_BUCKETING)
Expand All @@ -715,7 +697,6 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
.add(&COMPUTE_REPLICA_EXPIRATION_OFFSET)
.add(&COMPUTE_APPLY_COLUMN_DEMANDS)
.add(&COMPUTE_FLAT_MAP_FUEL)
.add(&CONSOLIDATING_VEC_GROWTH_DAMPENER)
.add(&ENABLE_COMPUTE_RENDER_FUELED_AS_SPECIFIC_COLLECTION)
.add(&ENABLE_COMPUTE_LOGICAL_BACKPRESSURE)
.add(&COMPUTE_LOGICAL_BACKPRESSURE_MAX_RETAINED_CAPABILITIES)
Expand Down
165 changes: 47 additions & 118 deletions src/compute/benches/correction.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,9 @@
// 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 for the MV sink `Correction` buffer on hydration-style workloads,
//! i.e., workloads where the input catches up with the current time by passing through
//! many distinct timestamps.
//!
//! 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
Expand All @@ -30,8 +30,7 @@ use criterion::measurement::WallTime;
use criterion::{
BatchSize, BenchmarkGroup, BenchmarkId, Criterion, criterion_group, criterion_main,
};
use mz_compute::sink::correction::CorrectionV1;
use mz_compute::sink::correction_v2::CorrectionV2;
use mz_compute::sink::correction::Correction;
use mz_ore::metrics::MetricsRegistry;
use mz_persist_client::cfg::PersistConfig;
use mz_persist_client::metrics::{Metrics, SinkMetrics};
Expand All @@ -42,78 +41,13 @@ use timely::progress::Antichain;
const CHAIN_PROPORTIONALITY: f64 = 3.0;
/// Default value of `compute_correction_v2_chunk_size`.
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,
V2,
}

impl Version {
fn name(self) -> &'static str {
match self {
Self::V1 => "v1",
Self::V2 => "v2",
}
}
}

/// Local dispatch over the correction buffer implementations.
///
/// The production `Correction` enum only contains v1 and v2, so the bench carries its own.
enum Corr {
V1(CorrectionV1<Row>),
V2(CorrectionV2<Row>),
}

impl Corr {
fn insert(&mut self, updates: &mut Vec<(Row, Timestamp, Diff)>) {
match self {
Self::V1(c) => c.insert(updates),
Self::V2(c) => c.insert(updates),
}
}

fn insert_negated(&mut self, updates: &mut Vec<(Row, Timestamp, Diff)>) {
match self {
Self::V1(c) => c.insert_negated(updates),
Self::V2(c) => c.insert_negated(updates),
}
}

fn updates_before(
&mut self,
upper: &Antichain<Timestamp>,
) -> Box<dyn Iterator<Item = (Row, Timestamp, Diff)> + '_> {
match self {
Self::V1(c) => Box::new(c.updates_before(upper)),
Self::V2(c) => Box::new(c.updates_before(upper)),
}
}

fn advance_since(&mut self, since: Antichain<Timestamp>) {
match self {
Self::V1(c) => c.advance_since(since),
Self::V2(c) => c.advance_since(since),
}
}

fn consolidate_at_since(&mut self) {
match self {
Self::V1(c) => c.consolidate_at_since(),
Self::V2(c) => c.consolidate_at_since(),
}
}
}

/// The shape of the update stream fed into the correction buffer.
#[derive(Clone, Copy)]
enum Pattern {
Expand Down Expand Up @@ -145,22 +79,14 @@ fn sink_metrics() -> SinkMetrics {
metrics.sink.clone()
}

fn make_correction(version: Version, metrics: &SinkMetrics) -> Corr {
let worker_metrics = metrics.for_worker(0);
match version {
Version::V1 => Corr::V1(CorrectionV1::new(
metrics.clone(),
worker_metrics,
GROWTH_DAMPENER,
)),
Version::V2 => Corr::V2(CorrectionV2::new(
metrics.clone(),
worker_metrics,
None,
CHAIN_PROPORTIONALITY,
CHUNK_SIZE,
)),
}
fn make_correction(metrics: &SinkMetrics) -> Correction<Row> {
Correction::with_params(
metrics.clone(),
metrics.for_worker(0),
None,
CHAIN_PROPORTIONALITY,
CHUNK_SIZE,
)
}

fn row(key: u64, value: u64) -> Row {
Expand Down Expand Up @@ -219,28 +145,35 @@ fn make_batches(num_ts: u64, pattern: Pattern) -> Vec<Vec<(Row, Timestamp, Diff)
/// Fill a fresh correction buffer with all batches, mimicking desired input that has
/// run far ahead of persist writes.
fn filled_correction(
version: Version,
metrics: &SinkMetrics,
batches: &[Vec<(Row, Timestamp, Diff)>],
) -> Corr {
let mut correction = make_correction(version, metrics);
) -> Correction<Row> {
let mut correction = make_correction(metrics);
for batch in batches {
correction.insert(&mut batch.clone());
}
correction
}

impl Corr {
/// Benchmark helpers modelling the `write_batches` operator's use of the buffer.
///
/// A trait rather than an inherent `impl`, because `Correction` is defined in another crate.
trait WriteSteps {
/// Emulate one `write_batches` write: read the updates before `upper` and feed back their
/// negations, like the persist input does.
fn write_step(&mut self, upper: &Antichain<Timestamp>) -> usize;
/// Read and count the updates before `upper`, without persist feedback.
fn read_count(&mut self, upper: &Antichain<Timestamp>) -> usize;
}

impl WriteSteps for Correction<Row> {
fn write_step(&mut self, upper: &Antichain<Timestamp>) -> usize {
let mut written: Vec<_> = self.updates_before(upper).collect();
let count = written.len();
self.insert_negated(&mut written);
count
}

/// Read and count the updates before `upper`, without persist feedback.
fn read_count(&mut self, upper: &Antichain<Timestamp>) -> usize {
self.updates_before(upper).count()
}
Expand All @@ -253,7 +186,7 @@ impl Corr {
/// updates come back through the persist input and are removed by `insert_negated`. We
/// model that feedback here, otherwise the buffer re-emits all previous updates on every
/// step and clone costs drown out the structure-management costs we want to measure.
fn drain_stepwise(mut correction: Corr, num_ts: u64) -> Corr {
fn drain_stepwise(mut correction: Correction<Row>, num_ts: u64) -> Correction<Row> {
for t in 0..num_ts {
let upper = Antichain::from_elem(Timestamp::from(t + 1));
let count = correction.write_step(&upper);
Expand All @@ -265,7 +198,7 @@ fn drain_stepwise(mut correction: Corr, num_ts: u64) -> Corr {

/// Advance the since across all buffered times at once, then consolidate and drain.
/// This exercises `Cursor::advance_by` with many distinct times below the since.
fn advance_jump(mut correction: Corr, num_ts: u64) -> Corr {
fn advance_jump(mut correction: Correction<Row>, num_ts: u64) -> Correction<Row> {
correction.advance_since(Antichain::from_elem(Timestamp::from(num_ts)));
correction.consolidate_at_since();
let upper = Antichain::from_elem(Timestamp::from(num_ts + 1));
Expand All @@ -281,23 +214,21 @@ fn bench_scenario<F>(
num_ts: u64,
routine: F,
) where
F: Fn(Corr, u64) -> Corr,
F: Fn(Correction<Row>, u64) -> Correction<Row>,
{
let batches = make_batches(num_ts, pattern);
for version in [Version::V1, Version::V2] {
group.bench_function(BenchmarkId::new(version.name(), num_ts), |b| {
b.iter_batched(
|| filled_correction(version, metrics, &batches),
|correction| routine(correction, num_ts),
BatchSize::PerIteration,
)
});
}
group.bench_function(BenchmarkId::from_parameter(num_ts), |b| {
b.iter_batched(
|| filled_correction(metrics, &batches),
|correction| routine(correction, num_ts),
BatchSize::PerIteration,
)
});
}

/// Configure a benchmark group for short wall-clock time.
///
/// The interesting effects here are order-of-magnitude differences between implementations, so we
/// The interesting effects here are order-of-magnitude differences across problem sizes, so we
/// trade statistical rigor for execution speed.
fn configure(group: &mut BenchmarkGroup<WallTime>) {
group
Expand Down Expand Up @@ -329,20 +260,18 @@ fn bench_correction(c: &mut Criterion) {
configure(&mut group);
for num_ts in num_ts_values {
let batches = make_batches(num_ts, pattern);
for version in [Version::V1, Version::V2] {
group.bench_function(BenchmarkId::new(version.name(), num_ts), |b| {
b.iter_batched(
|| (make_correction(version, &metrics), batches.clone()),
|(mut correction, mut batches)| {
for batch in &mut batches {
correction.insert(batch);
}
correction
},
BatchSize::PerIteration,
)
});
}
group.bench_function(BenchmarkId::from_parameter(num_ts), |b| {
b.iter_batched(
|| (make_correction(&metrics), batches.clone()),
|(mut correction, mut batches)| {
for batch in &mut batches {
correction.insert(batch);
}
correction
},
BatchSize::PerIteration,
)
});
}
group.finish();
}
Expand Down
4 changes: 0 additions & 4 deletions src/compute/src/sink.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,6 @@ mod copy_to_s3_oneshot;
pub mod correction;
#[cfg(not(feature = "bench"))]
mod correction;
#[cfg(feature = "bench")]
pub mod correction_v2;
#[cfg(not(feature = "bench"))]
mod correction_v2;
mod materialized_view;
mod materialized_view_v2;
mod metric_sink;
Expand Down
Loading
Loading