From 0ff833a77c686f4375743a504ed548c8da8b654f Mon Sep 17 00:00:00 2001 From: Ben Kirwin Date: Tue, 10 Mar 2026 17:45:05 -0400 Subject: [PATCH 1/3] persist: a shared batch builder for coalescing writes across workers `SharedBatches` hands out a `SharedBatchBuilder` per batch id. Handles for the same id feed one process-global builder task, so the workers running in one process contribute their parts to a single, larger batch instead of each writing its own small one. The last handle to `finish` receives the batch; the others receive `None`. `BatchBuilder::add_part` accepts a pre-encoded part and concatenates parts until they reach the blob target size, which is what lets the shared builder accept work from several workers without re-encoding it. Building a batch now goes through `PersistClient::batch_builder` rather than a `WriteHandle`, so a shared builder needs no writer registration of its own. Co-authored-by: Moritz Hoffmann --- src/compute/src/compute_state.rs | 6 + src/compute/src/compute_state/peek_stash.rs | 15 +- src/compute/src/server.rs | 7 + src/persist-client/src/batch.rs | 132 +++++++-- src/persist-client/src/lib.rs | 5 +- src/persist-types/src/part.rs | 5 + src/storage-operators/Cargo.toml | 1 + src/storage-operators/src/lib.rs | 2 + src/storage-operators/src/persist.rs | 305 ++++++++++++++++++++ 9 files changed, 441 insertions(+), 37 deletions(-) create mode 100644 src/storage-operators/src/persist.rs diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index 1569592ede151..f6d197eb9d823 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -52,6 +52,7 @@ use mz_persist_types::PersistLocation; use mz_persist_types::codec_impls::UnitSchema; use mz_repr::fixed_length::ExtendDatums; use mz_repr::{DatumVec, Diff, GlobalId, Row, RowArena, Timestamp}; +use mz_storage_operators::persist::SharedBatches; use mz_storage_operators::stats::StatsCursor; use mz_storage_types::StorageDiff; use mz_storage_types::controller::CollectionMetadata; @@ -114,6 +115,9 @@ pub struct ComputeState { /// A process-global cache of (blob_uri, consensus_uri) -> PersistClient. /// This is intentionally shared between workers. pub persist_clients: Arc, + // A process-global cache of shared persist batch writers. This allows + // coalescing writes across local workers without explicit exchanges. + pub persist_batches: SharedBatches, /// Context necessary for rendering txn-wal operators. pub txns_ctx: TxnsContext, /// History of commands received by this workers and all its peers. @@ -178,6 +182,7 @@ impl ComputeState { /// Construct a new `ComputeState`. pub fn new( persist_clients: Arc, + persist_batches: SharedBatches, txns_ctx: TxnsContext, metrics: WorkerMetrics, tracing_handle: Arc, @@ -198,6 +203,7 @@ impl ComputeState { peek_stash_persist_location: None, compute_logger: None, persist_clients, + persist_batches, txns_ctx, command_history, max_result_size: u64::MAX, diff --git a/src/compute/src/compute_state/peek_stash.rs b/src/compute/src/compute_state/peek_stash.rs index 214d9f371f64f..95a24dc689358 100644 --- a/src/compute/src/compute_state/peek_stash.rs +++ b/src/compute/src/compute_state/peek_stash.rs @@ -155,14 +155,13 @@ impl StashingPeek { // // TODO: We _could_ work around the above by teaching the bare columnar // Row encoder about zero-column rows. - let mut batch_builder = client - .batch_builder::( - shard_id, - write_schemas, - lower, - Some(batch_max_runs), - ) - .await; + let mut batch_builder = client.batch_builder::( + shard_id, + "peek_stash", + write_schemas, + lower, + Some(batch_max_runs), + ); let mut num_rows: u64 = 0; diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs index 3fdbd78e0060c..2eeaafca6abe1 100644 --- a/src/compute/src/server.rs +++ b/src/compute/src/server.rs @@ -29,6 +29,7 @@ use mz_ore::halt; use mz_ore::metrics::MetricsRegistry; use mz_ore::tracing::TracingHandle; use mz_persist_client::cache::PersistClientCache; +use mz_storage_operators::persist::SharedBatches; use mz_storage_types::connections::ConnectionContext; use mz_timely_util::capture::EventLink; use mz_txn_wal::operator::TxnsContext; @@ -129,6 +130,8 @@ pub(crate) type StorageTimelyLogReader = struct Config { /// `persist` client cache. pub persist_clients: Arc, + /// Shared persist batch state. + pub persist_batches: SharedBatches, /// Context necessary for rendering txn-wal operators. pub txns_ctx: TxnsContext, /// A process-global handle to tracing configuration. @@ -173,6 +176,7 @@ pub async fn serve( mz_timely_util::pool_config::metrics::register(metrics_registry); let config = Config { + persist_batches: SharedBatches::new(), persist_clients, txns_ctx, tracing_handle, @@ -305,6 +309,7 @@ struct Worker<'w> { /// A process-global cache of (blob_uri, consensus_uri) -> PersistClient. /// This is intentionally shared between workers persist_clients: Arc, + persist_batches: SharedBatches, /// Context necessary for rendering txn-wal operators. txns_ctx: TxnsContext, /// A process-global handle to tracing configuration. @@ -361,6 +366,7 @@ impl ClusterSpec for Config { metrics, context: self.context.clone(), persist_clients: Arc::clone(&self.persist_clients), + persist_batches: self.persist_batches.clone(), txns_ctx: self.txns_ctx.clone(), compute_state: None, tracing_handle: Arc::clone(&self.tracing_handle), @@ -498,6 +504,7 @@ impl<'w> Worker<'w> { if matches!(&cmd, ComputeCommand::CreateInstance(_)) { self.compute_state = Some(ComputeState::new( Arc::clone(&self.persist_clients), + self.persist_batches.clone(), self.txns_ctx.clone(), self.metrics.clone(), Arc::clone(&self.tracing_handle), diff --git a/src/persist-client/src/batch.rs b/src/persist-client/src/batch.rs index 3c80d5e5084a9..1bc752b2f990a 100644 --- a/src/persist-client/src/batch.rs +++ b/src/persist-client/src/batch.rs @@ -30,7 +30,7 @@ use mz_ore::instrument; use mz_persist::indexed::encoding::{BatchColumnarFormat, BlobTraceBatchPart, BlobTraceUpdates}; use mz_persist::location::Blob; use mz_persist_types::arrow::{ArrayBound, ArrayOrd}; -use mz_persist_types::columnar::{ColumnDecoder, Schema}; +use mz_persist_types::columnar::{ColumnDecoder, Schema, data_type}; use mz_persist_types::parquet::{CompressionFormat, EncodingConfig}; use mz_persist_types::part::{Part, PartBuilder}; use mz_persist_types::schema::SchemaId; @@ -535,7 +535,15 @@ where inline_desc: Description, inclusive_upper: Antichain>, - records_builder: PartBuilder, + part_builder: PartBuilder, + finished_parts: Vec, + /// The accumulated size of `finished_parts`. + /// + /// Tracked incrementally because `Part::goodbytes` rebuilds an `ArrayOrd` per key and value + /// column on every call, and parts accumulate until they reach the blob target size. Summing + /// the vector on each `add`/`add_part` would make a flush window quadratic in the number of + /// parts it holds. + finished_parts_goodbytes: usize, pub(crate) builder: BatchBuilderInternal, } @@ -557,11 +565,46 @@ where Self { inline_desc, inclusive_upper: Antichain::new(), - records_builder, + part_builder: records_builder, + finished_parts: vec![], + finished_parts_goodbytes: 0, builder, } } + fn in_progress_goodbytes(&self) -> usize { + self.part_builder.goodbytes() + self.finished_parts_goodbytes + } + + fn push_finished_part(&mut self, part: Part) { + self.finished_parts_goodbytes += part.goodbytes(); + self.finished_parts.push(part); + } + + fn finish_part_builder(&mut self) { + if self.part_builder.len() > 0 { + let part = self.part_builder.finish_and_replace( + &*self.builder.write_schemas.key, + &*self.builder.write_schemas.val, + ); + self.push_finished_part(part); + } + } + + async fn flush_parts(&mut self) -> bool { + self.finish_part_builder(); + if let Some(part) = Part::concat(&self.finished_parts).expect("type aligned") { + self.builder + .flush_part(self.inline_desc.clone(), part) + .await; + self.finished_parts.clear(); + self.finished_parts_goodbytes = 0; + true + } else { + false + } + } + /// Finish writing this batch and return a handle to the written batch. /// /// This fails if any of the updates in this batch are beyond the given @@ -592,10 +635,7 @@ where } } - let updates = self.records_builder.finish(); - self.builder - .flush_part(self.inline_desc.clone(), updates) - .await; + self.flush_parts().await; self.builder .finish(Description::new( @@ -606,6 +646,56 @@ where .await } + /// Adds the given part to the batch, encoded with this builder's write schemas. + /// + /// Every update timestamp in the part must be greater or equal to `lower` that was given when + /// creating this [BatchBuilder]. + /// + /// Parts accumulate until they reach the configured blob target size and are then concatenated + /// into one flushed part, so an individual `part` must stay well below persist's + /// `KEY_VAL_DATA_MAX_LEN`: a single oversized part is flushed on its own, but nothing splits + /// it. + pub async fn add_part(&mut self, part: Part) -> Result> { + for time in part.time.values() { + let ts = T::decode(time.to_le_bytes()); + if !self.inline_desc.lower().less_equal(&ts) { + return Err(InvalidUsage::UpdateNotBeyondLower { + ts, + lower: self.inline_desc.lower().clone(), + }); + } + self.inclusive_upper.insert(Reverse(ts)); + } + + assert_eq!( + data_type::(&self.builder.write_schemas.key).expect("valid type"), + *part.key.data_type(), + ); + assert_eq!( + data_type::(&self.builder.write_schemas.val).expect("valid type"), + *part.val.data_type() + ); + + self.finish_part_builder(); + + let mut flushed = false; + if self.in_progress_goodbytes() + part.goodbytes() > self.builder.parts.cfg.blob_target_size + { + flushed |= self.flush_parts().await; + } + self.push_finished_part(part); + if self.in_progress_goodbytes() > self.builder.parts.cfg.blob_target_size { + flushed |= self.flush_parts().await; + } + + let added = if flushed { + Added::RecordAndParts + } else { + Added::Record + }; + Ok(added) + } + /// Adds the given update to the batch. /// /// The update timestamp must be greater or equal to `lower` that was given @@ -625,24 +715,13 @@ where } self.inclusive_upper.insert(Reverse(ts.clone())); - let added = { - self.records_builder - .push(key, val, ts.clone(), diff.clone()); - if self.records_builder.goodbytes() >= self.builder.parts.cfg.blob_target_size { - let part = self.records_builder.finish_and_replace( - self.builder.write_schemas.key.as_ref(), - self.builder.write_schemas.val.as_ref(), - ); - Some(part) - } else { - None - } - }; + let mut flushed = false; + self.part_builder.push(key, val, ts.clone(), diff.clone()); + if self.in_progress_goodbytes() >= self.builder.parts.cfg.blob_target_size { + flushed |= self.flush_parts().await; + } - let added = if let Some(full_batch) = added { - self.builder - .flush_part(self.inline_desc.clone(), full_batch) - .await; + let added = if flushed { Added::RecordAndParts } else { Added::Record @@ -763,9 +842,8 @@ where Ok(batch) } - /// Flushes the current part to Blob storage, first consolidating and then - /// columnar encoding the updates. It is the caller's responsibility to - /// chunk `current_part` to be no greater than + /// Flushes the current part to Blob storage, first columnar encoding the updates. + /// It is the caller's responsibility to chunk `current_part` to be no greater than /// [BatchBuilderConfig::blob_target_size], and must absolutely be less than /// [mz_persist::indexed::columnar::KEY_VAL_DATA_MAX_LEN] pub async fn flush_part(&mut self, part_desc: Description, columnar: Part) { diff --git a/src/persist-client/src/lib.rs b/src/persist-client/src/lib.rs index 1c22327b27d2e..edc786a03420a 100644 --- a/src/persist-client/src/lib.rs +++ b/src/persist-client/src/lib.rs @@ -551,9 +551,10 @@ impl PersistClient { /// enough that we can reasonably chunk them up: O(KB) is definitely fine, /// O(MB) come talk to us. #[instrument(level = "debug", fields(shard = %shard_id))] - pub async fn batch_builder( + pub fn batch_builder( &self, shard_id: ShardId, + shard_name: &str, write_schemas: Schemas, lower: Antichain, max_runs: Option, @@ -570,7 +571,7 @@ impl PersistClient { &self.cfg, compact_cfg, Arc::clone(&self.metrics), - self.metrics.shards.shard(&shard_id, "peek_stash"), + self.metrics.shards.shard(&shard_id, shard_name), &self.metrics.user, Arc::clone(&self.isolated_runtime), Arc::clone(&self.blob), diff --git a/src/persist-types/src/part.rs b/src/persist-types/src/part.rs index 2c932d22bed02..0ad2f419cd952 100644 --- a/src/persist-types/src/part.rs +++ b/src/persist-types/src/part.rs @@ -187,6 +187,11 @@ impl, V, VS: Schema> PartBuilder { } } + /// The number of elements pushed into the builder. + pub fn len(&self) -> usize { + self.time.len() + } + /// Estimate the size of the part this builder will build. pub fn goodbytes(&self) -> usize { self.key.goodbytes() + self.val.goodbytes() + self.time.goodbytes() + self.diff.goodbytes() diff --git a/src/storage-operators/Cargo.toml b/src/storage-operators/Cargo.toml index 80938c15f2bc8..36b708cc76ca2 100644 --- a/src/storage-operators/Cargo.toml +++ b/src/storage-operators/Cargo.toml @@ -51,6 +51,7 @@ thiserror.workspace = true tokio.workspace = true tokio-util = { workspace = true, features = ["io"] } tracing.workspace = true +uuid = { workspace = true, features = ["v4"] } [dev-dependencies] bytesize.workspace = true diff --git a/src/storage-operators/src/lib.rs b/src/storage-operators/src/lib.rs index 8a78eed1605e6..76e6c6df8bc70 100644 --- a/src/storage-operators/src/lib.rs +++ b/src/storage-operators/src/lib.rs @@ -14,3 +14,5 @@ pub mod oneshot_source; pub mod persist_source; pub mod s3_oneshot_sink; pub mod stats; + +pub mod persist; diff --git a/src/storage-operators/src/persist.rs b/src/storage-operators/src/persist.rs new file mode 100644 index 0000000000000..3bedb13fec9ed --- /dev/null +++ b/src/storage-operators/src/persist.rs @@ -0,0 +1,305 @@ +// 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::task::JoinHandle; +use mz_persist_client::batch::Batch; +use mz_persist_client::{PersistClient, Schemas}; +use mz_persist_types::ShardId; +use mz_persist_types::part::Part; +use mz_repr::Timestamp; +use mz_storage_types::StorageDiff; +use mz_storage_types::sources::SourceData; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; +use std::fmt::Debug; +use std::sync::{Arc, Weak}; +use timely::progress::Antichain; +use uuid::Uuid; + +/// A unique identifier for a [SharedBatchBuilder]. +#[derive( + Debug, + Copy, + Clone, + Serialize, + Deserialize, + Ord, + PartialOrd, + Eq, + PartialEq, + Hash +)] +#[serde(transparent)] +pub struct SharedBatchId(Uuid); + +impl SharedBatchId { + pub fn new() -> Self { + Self(Uuid::new_v4()) + } +} + +/// A struct from which to obtain a [SharedBatchBuilder]. +#[derive(Debug, Clone)] +pub struct SharedBatches { + data: Arc>)>>, +} + +fn upgrade_or_init(weak: &mut Weak, init: impl FnOnce() -> T) -> Arc { + if let Some(owned) = weak.upgrade() { + owned + } else { + let owned = Arc::new(init()); + *weak = Arc::downgrade(&owned); + owned + } +} + +impl SharedBatches { + pub fn new() -> Self { + Self { + data: Arc::default(), + } + } + + /// Get a builder for the specified batch. If there is an existing, live builder for the same + /// batch id managed by [Self], any parts added to this builder will be included in the same, + /// shared batch; if not, a new batch will be created. + /// + /// This is designed so that, in a multi-worker dataflow, workers that are building the same batch + /// at the same time can write fewer, larger parts instead of many small ones. It does not guarantee + /// that there will be _precisely_ one batch across all workers, however. + /// + /// The batch's description is fixed by whichever caller creates it, so every caller sharing a + /// `shared_batch_id` must pass the same `shard_id`, `lower`, and `upper`. Callers derive them + /// from a single broadcast description, and the assertions below hold them to it: a caller + /// whose description disagreed would otherwise have its updates silently written under someone + /// else's bounds. + pub fn builder( + &self, + shared_batch_id: SharedBatchId, + client: PersistClient, + shard_id: ShardId, + shard_name: String, + schemas: Schemas, + lower: Antichain, + upper: Antichain, + ) -> SharedBatchBuilder { + let mut guard = self.data.lock().unwrap(); + let (last_retained_len, state_map) = &mut *guard; + // We clean out entries from the map where all shared batch handles have been dropped. + // Amortize by only scanning the map once it's doubled in size. + if state_map.len() > *last_retained_len * 2 { + // Probe the count rather than `upgrade()`: the temporary strong reference an upgrade + // creates can make a concurrent `SharedBatchBuilder::finish` see a strong count above + // one and hand the batch to nobody, which seals the interval with no data. A stale + // count is harmless in both directions, since zero cannot become live again and a + // stale non-zero only keeps a dead entry until the next sweep. + state_map.retain(|_, weak| weak.strong_count() > 0); + *last_retained_len = state_map.len(); + } + let weak = state_map.entry(shared_batch_id).or_default(); + let desc = BatchDesc { + shard_id, + lower: lower.clone(), + upper: upper.clone(), + }; + let state = upgrade_or_init(weak, || { + let (tx, mut rx) = tokio::sync::mpsc::channel(4); + let task_id = format!("shared-batch-{shared_batch_id:?}-{shard_id}",); + BatchState { + desc: desc.clone(), + tx, + handle: mz_ore::task::spawn(|| task_id, async move { + let mut builder = None; + while let Some(cmd) = rx.recv().await { + match cmd { + BatchCommand::Push(part) => { + let builder = builder.get_or_insert_with(|| { + client.batch_builder( + shard_id, + &shard_name, + schemas.clone(), + lower.clone(), + None, + ) + }); + builder.add_part(part).await.expect("valid timestamps"); + } + } + } + if let Some(builder) = builder { + Some(builder.finish(upper).await.expect("valid upper bound")) + } else { + None + } + }), + } + }); + + assert_eq!( + state.desc, desc, + "shared batch {shared_batch_id:?} requested with conflicting descriptions", + ); + + SharedBatchBuilder { + batch_id: shared_batch_id, + shard_id, + state, + } + } +} + +/// The bounds a shared batch is built under, fixed by the caller that creates it. +#[derive(Debug, Clone, PartialEq, Eq)] +struct BatchDesc { + shard_id: ShardId, + lower: Antichain, + upper: Antichain, +} + +enum BatchCommand { + Push(Part), +} + +#[derive(Debug)] +struct BatchState { + desc: BatchDesc, + tx: tokio::sync::mpsc::Sender, + handle: JoinHandle>>, +} + +/// A handle for a shared batch builder. Everyone with a handle for the same builder can +/// [Self::push] to that builder, but the last handle to call [Self::finish] will receive a batch +/// containing all the data. +/// +/// Note that it's quite important to call [Self::finish], even if you haven't pushed anything +/// into the batch, in case you're holding the last builder for a shared batch. +/// +/// Dropping the last handle instead still completes the batch, but hands it to nobody, so its +/// parts stay in blob until persist's leaked-blob cleanup collects them. That is the teardown +/// path, not a path any caller should take deliberately. +pub struct SharedBatchBuilder { + batch_id: SharedBatchId, + shard_id: ShardId, + state: Arc, +} + +impl Debug for SharedBatchBuilder { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SharedBatchBuilder") + .field("batch_id", &self.batch_id) + .field("shard_id", &self.shard_id) + .finish_non_exhaustive() + } +} + +impl SharedBatchBuilder { + /// Include the provided part in the batch. If there are a large number of updates that have + /// not been flushed to S3, this call may wait. + pub async fn push(&self, part: Part) { + if part.len() == 0 { + return; + } + self.state + .tx + .send(BatchCommand::Push(part)) + .await + .expect("task failed"); + } + + /// Fetch the results of the batch builder from the shared state, if any. + /// + /// Only the last builder to call this method will obtain the resulting batch, and that batch + /// will include all data written by all workers. + /// This means that, for any batch interval that we actually want to append, all workers + /// must call finish even if they did not push any batches themselves. + pub async fn finish(self) -> Option> { + let state = Arc::into_inner(self.state)?; + drop(state.tx); // The task only finishes the batch once the channel is closed. + state.handle.await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use mz_persist_types::codec_impls::UnitSchema; + use mz_persist_types::part::PartBuilder; + use mz_repr::{Datum, RelationDesc, Row, SqlScalarType}; + + fn bool_schemas() -> Schemas { + Schemas { + id: None, + key: Arc::new( + RelationDesc::builder() + .with_column("test", SqlScalarType::Bool.nullable(true)) + .finish(), + ), + val: Arc::new(UnitSchema), + } + } + + fn bool_part(schemas: &Schemas) -> Part { + let mut builder = PartBuilder::new(&*schemas.key, &*schemas.val); + builder.push( + &SourceData(Ok(Row::pack_slice(&[Datum::True]))), + &(), + Timestamp::new(0), + 1i64, + ); + builder.finish() + } + + #[mz_ore::test(tokio::test)] + async fn test_shared_batch() { + let client = PersistClient::new_for_tests().await; + let shared = SharedBatches::new(); + let batch_id = SharedBatchId::new(); + let shard_id = ShardId::new(); + let schemas = bool_schemas(); + let lower = Antichain::from_elem(0.into()); + let upper = Antichain::from_elem(1.into()); + let first = shared.builder( + batch_id, + client.clone(), + shard_id, + "test".to_string(), + schemas.clone(), + lower.clone(), + upper.clone(), + ); + let second = shared.builder( + batch_id, + client, + shard_id, + "test".to_string(), + schemas.clone(), + lower.clone(), + upper.clone(), + ); + let part = bool_part(&schemas); + first.push(part.clone()).await; + second.push(part.clone()).await; + assert!( + second.finish().await.is_none(), + "a handle that is not the last to finish receives no batch" + ); + let batch = first.finish().await.unwrap(); + assert_eq!( + batch.shard_id(), + shard_id, + "batch should be for the expected shard" + ); + assert_eq!( + batch.into_hollow_batch().len, + 2, + "batch should include updates from both pushes" + ) + } +} From ae3b03bc7c5551fe1accf38c305b8d91cd4c0019 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Sun, 12 Jul 2026 02:40:44 +0200 Subject: [PATCH 2/3] compute: coalesce MV sink batches across workers in the sync sink Wire the shared-batch builder into the sync (v2) materialized-view sink so the workers in one process coalesce their parts for a given batch interval into a single, larger batch instead of each writing its own small batch. All workers building an interval share the batch description broadcast by the mint operator, so they share a batch id and their parts land in one process-global shared batch. Only the last worker to finish receives that batch; the rest get nothing, which maps to the empty-batch response the write operator already handles. Every worker still finishes its builder even when it pushed no data, since any one of them may be the last holder responsible for delivering all workers' parts. The behavior is gated behind the new enable_compute_sync_mv_sink_shared_batches dyncfg, default off. The prior per-worker path is retained unchanged for the off case. Measured locally on an 8-worker replica with a churning view: blob PUTs down ~67% and consensus state-diff bytes down ~54%, at unchanged consensus command count. Adds a testdrive test asserting the view equals the equivalent one-shot aggregation across a multi-worker cluster with churn, enables the flag in the CI system-parameter defaults, and registers it with parallel-workload's flag flipper. Co-Authored-By: Claude Opus 4.8 (1M context) --- misc/python/materialize/mzcompose/__init__.py | 5 + .../materialize/parallel_workload/action.py | 3 + src/compute-types/src/dyncfgs.rs | 11 ++ src/compute/src/sink/materialized_view.rs | 6 + src/compute/src/sink/materialized_view_v2.rs | 160 +++++++++++++++--- test/testdrive/mv-sink-shared-batches.td | 81 +++++++++ 6 files changed, 240 insertions(+), 26 deletions(-) create mode 100644 test/testdrive/mv-sink-shared-batches.td diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 9ce47a13dc0a5..3dcc2f853679a 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -305,6 +305,11 @@ def get_variable_system_parameters( "true", ["true", "false"], ), + VariableSystemParameter( + "enable_compute_sync_mv_sink_shared_batches", + "true", + ["true", "false"], + ), VariableSystemParameter( "enable_password_auth", "true", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index a64d4b3b3a094..618aa4fe65914 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -2882,6 +2882,9 @@ def __init__( BOOLEAN_FLAG_VALUES ) self.flags_with_values["enable_eager_delta_joins"] = BOOLEAN_FLAG_VALUES + self.flags_with_values["enable_compute_sync_mv_sink_shared_batches"] = ( + BOOLEAN_FLAG_VALUES + ) self.flags_with_values["enable_public_metrics_endpoint"] = BOOLEAN_FLAG_VALUES self.flags_with_values["persist_batch_structured_key_lower_len"] = [ "0", diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index 2f7a2c7586549..fb891a3d1d687 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -262,6 +262,16 @@ pub const ENABLE_SYNC_MV_SINK: Config = Config::new( ParameterScope::Environment, ); +/// Coalesce the parts written by the workers in one process into a single shared batch per batch +/// interval, instead of each worker writing its own batch. +pub const ENABLE_SYNC_MV_SINK_SHARED_BATCHES: Config = Config::new( + "enable_compute_sync_mv_sink_shared_batches", + false, + "Coalesce the parts written by the workers in one process into a single shared batch per \ + batch interval in the MV sink.", + ParameterScope::Environment, +); + /// Whether rendering should use the new MV sink correction buffer implementation. pub const ENABLE_CORRECTION_V2: Config = Config::new( "enable_compute_correction_v2", @@ -688,6 +698,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&ENABLE_ERROR_DISTINCT) .add(&ENABLE_MZ_JOIN_CORE) .add(&ENABLE_SYNC_MV_SINK) + .add(&ENABLE_SYNC_MV_SINK_SHARED_BATCHES) .add(&ENABLE_CORRECTION_V2) .add(&CORRECTION_V2_CHAIN_PROPORTIONALITY) .add(&CORRECTION_V2_CHUNK_SIZE) diff --git a/src/compute/src/sink/materialized_view.rs b/src/compute/src/sink/materialized_view.rs index e5d1a4064c1ba..6c193fd1280b6 100644 --- a/src/compute/src/sink/materialized_view.rs +++ b/src/compute/src/sink/materialized_view.rs @@ -134,6 +134,7 @@ use mz_persist_client::write::WriteHandle; use mz_persist_client::{Diagnostics, PersistClient}; use mz_persist_types::codec_impls::UnitSchema; use mz_repr::{Diff, GlobalId, Row, Timestamp}; +use mz_storage_operators::persist::{SharedBatchId, SharedBatches}; use mz_storage_types::StorageDiff; use mz_storage_types::controller::CollectionMetadata; use mz_storage_types::sources::SourceData; @@ -267,6 +268,7 @@ where let persist_api = PersistApi { persist_clients: Arc::clone(&compute_state.persist_clients), + persist_batches: compute_state.persist_batches.clone(), collection: target.clone(), shard_name: sink_id.to_string(), purpose: format!("MV sink {sink_id}"), @@ -351,6 +353,7 @@ pub(super) fn advance( #[derive(Clone)] pub(super) struct PersistApi { pub(super) persist_clients: Arc, + pub(super) persist_batches: SharedBatches, pub(super) collection: CollectionMetadata, pub(super) shard_name: String, pub(super) purpose: String, @@ -442,6 +445,8 @@ pub(super) struct BatchDescription { pub(super) lower: Antichain, pub(super) upper: Antichain, pub(super) append_worker: usize, + /// Identifies the shared batch all workers building this description contribute to. + pub(super) shared_id: SharedBatchId, } impl BatchDescription { @@ -455,6 +460,7 @@ impl BatchDescription { lower, upper, append_worker, + shared_id: SharedBatchId::new(), } } } diff --git a/src/compute/src/sink/materialized_view_v2.rs b/src/compute/src/sink/materialized_view_v2.rs index 5408d63142025..7430f669f9169 100644 --- a/src/compute/src/sink/materialized_view_v2.rs +++ b/src/compute/src/sink/materialized_view_v2.rs @@ -52,12 +52,19 @@ use std::rc::Rc; use std::sync::Arc; use differential_dataflow::{Hashable, VecCollection}; -use mz_compute_types::dyncfgs::MV_SINK_ADVANCE_PERSIST_FRONTIERS; +use mz_compute_types::dyncfgs::{ + ENABLE_SYNC_MV_SINK_SHARED_BATCHES, MV_SINK_ADVANCE_PERSIST_FRONTIERS, +}; use mz_dyncfg::ConfigSet; use mz_ore::cast::CastFrom; use mz_persist_client::batch::{Batch, ProtoBatch}; use mz_persist_client::write::WriteHandle; +use mz_persist_client::{PersistClient, Schemas}; +use mz_persist_types::ShardId; +use mz_persist_types::codec_impls::UnitSchema; +use mz_persist_types::part::PartBuilder; use mz_repr::{Diff, GlobalId, Row, Timestamp}; +use mz_storage_operators::persist::SharedBatches; use mz_storage_types::StorageDiff; use mz_storage_types::sources::SourceData; use timely::PartialOrder; @@ -102,6 +109,7 @@ pub(super) fn persist_sink<'s>( let persist_api = PersistApi { persist_clients: Arc::clone(&compute_state.persist_clients), + persist_batches: compute_state.persist_batches.clone(), collection: target.clone(), shard_name: sink_id.to_string(), purpose: format!("MV sink {sink_id}"), @@ -655,6 +663,10 @@ mod write { let advance_persist_frontiers_at_startup = MV_SINK_ADVANCE_PERSIST_FRONTIERS.get(&worker_config); + // Read the shared-batch flag on the Timely thread. `worker_config` is an `Rc` + // and thus not `Send`, so the value must be captured before it moves into the Tokio task. + let shared_batches_enabled = ENABLE_SYNC_MV_SINK_SHARED_BATCHES.get(&worker_config); + // Mirror the persist-frontier initialization performed by `State::new` below. With the // flag enabled, `State` advances its Timely-side `persist_frontiers` to `as_of`, opening // the `maybe_start_batch` write gate (`desc.lower <= persist_frontiers.frontier()`) @@ -679,17 +691,17 @@ mod write { let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::(); let (resp_tx, mut resp_rx) = mpsc::unbounded_channel::(); - // Spawn Tokio task that owns the WriteHandle and corrections buffer. + // Spawn Tokio task that owns the batch-writing context and corrections buffer. let (activator, activation_ack) = ArcActivator::new(scope, &info); let write_task_handle = { mz_ore::task::spawn( || operator_name(sink_id, "write::batch_writer"), async move { - let writer = persist_api.open_writer().await; + let batch_writer = BatchWriter::new(&persist_api, shared_batches_enabled).await; while let Some(cmd) = cmd_rx.recv().await { corrections = - apply_command(sink_id, corrections, &writer, cmd, &resp_tx).await; + apply_command(sink_id, corrections, &batch_writer, cmd, &resp_tx).await; // Activate the operator to drain logging events and process batch responses. // ArcActivator suppresses redundant activations, so this is cheap. activator.activate(); @@ -841,6 +853,52 @@ mod write { /// batch builder. Together with [`READ_BACK_CHUNK`] this bounds the handoff to a few thousand /// buffered updates. const READ_BACK_CHUNKS_IN_FLIGHT: usize = 4; + /// How many updates a worker accumulates before handing a part to the shared batch. + /// + /// The shared batch concatenates the parts it receives until they reach persist's blob target + /// size, so this only sets the granularity at which a worker's updates become visible to that + /// coalescing, not the size of the parts persist writes. Keeping it near [`READ_BACK_CHUNK`] + /// costs one part per read-back chunk. + const SHARED_PART_LEN: usize = 1024; + + /// The batch-writing context owned by the Tokio write task. + /// + /// Batches are built through the process-global [`SharedBatches`], so that the workers running + /// in one process coalesce their parts for a given batch interval into a single, larger batch + /// instead of each writing its own small batch. The `writer` is retained only to clean up a + /// finished batch when the response channel has already gone away. + struct BatchWriter { + persist_client: PersistClient, + shard_id: ShardId, + shard_name: String, + schemas: Schemas, + shared_batches: SharedBatches, + shared_batches_enabled: bool, + writer: WriteHandle, + } + + impl BatchWriter { + async fn new(persist_api: &PersistApi, shared_batches_enabled: bool) -> Self { + let writer = persist_api.open_writer().await; + Self { + persist_client: persist_api.open_client().await, + shard_id: persist_api.collection.data_shard, + shard_name: persist_api.shard_name.clone(), + // Parts carry the id of the schema they were written under, and a part written + // under none cannot be decoded through the shard's schema registry. The handle + // resolves the registered id for this sink's schema, so take it from there rather + // than assembling an unregistered `Schemas`. + schemas: Schemas { + id: writer.schema_id(), + key: Arc::new(persist_api.collection.relation_desc.clone()), + val: Arc::new(UnitSchema), + }, + shared_batches: persist_api.persist_batches.clone(), + shared_batches_enabled, + writer, + } + } + } /// Apply a single command to the task state, returning the correction buffers. /// @@ -857,7 +915,7 @@ mod write { async fn apply_command( sink_id: GlobalId, mut corrections: Corrections, - writer: &WriteHandle, + batch_writer: &BatchWriter, cmd: WriteCommand, resp_tx: &mpsc::UnboundedSender, ) -> Corrections { @@ -914,30 +972,39 @@ mod write { }, ); - // Create the builder lazily: an idle sink's descriptions find no corrections. - let mut builder = None; - while let Some(chunk) = updates_rx.recv().await { - let builder = builder.get_or_insert_with(|| writer.builder(desc.lower.clone())); - for ((k, v), t, d) in &chunk { - builder.add(k, v, t, d).await.expect("valid usage"); + let proto_batch = if batch_writer.shared_batches_enabled { + write_shared_batch(batch_writer, &desc, &mut updates_rx).await + } else { + // Per-worker path: each worker writes its own batch for the interval. Create + // the builder lazily: an idle sink's descriptions find no corrections. + let mut builder = None; + while let Some(chunk) = updates_rx.recv().await { + let builder = builder + .get_or_insert_with(|| batch_writer.writer.builder(desc.lower.clone())); + for ((k, v), t, d) in &chunk { + builder.add(k, v, t, d).await.expect("valid usage"); + } + } + match builder { + Some(builder) => { + let batch = builder + .finish(desc.upper.clone()) + .await + .expect("valid usage"); + Some(batch.into_transmittable_batch()) + } + None => None, } - } - let corrections = read_back.await; - - let Some(builder) = builder else { - // No corrections to write. - let _ = resp_tx.send(WriteResponse { batch: None }); - return corrections; }; + let corrections = read_back.await; - let batch = builder.finish(desc.upper).await.expect("valid usage"); - let proto_batch = batch.into_transmittable_batch(); - if let Err(err) = resp_tx.send(WriteResponse { - batch: Some(proto_batch), - }) { - let batch = - writer.batch_from_transmittable_batch(err.0.batch.expect("just sent")); - batch.delete().await; + if let Err(err) = resp_tx.send(WriteResponse { batch: proto_batch }) { + if let Some(proto_batch) = err.0.batch { + let batch = batch_writer + .writer + .batch_from_transmittable_batch(proto_batch); + batch.delete().await; + } } corrections @@ -998,6 +1065,47 @@ mod write { .await } + /// Build the batch for `desc` through the process-global [`SharedBatches`], coalescing this + /// worker's parts with those of the other workers in the process that build the same interval. + /// + /// All workers building an interval share `desc.shared_id` (the `mint` operator broadcasts one + /// description), so their parts land in a single shared batch. Only the last worker to `finish` + /// receives that batch; the rest get `None`, which maps to the empty-batch response the write + /// operator already handles. Every worker must `finish` even when it pushed nothing, since any + /// one of them may be the last holder and thus responsible for delivering all workers' data. + async fn write_shared_batch( + batch_writer: &BatchWriter, + desc: &BatchDescription, + updates_rx: &mut mpsc::Receiver>, + ) -> Option { + let schemas = &batch_writer.schemas; + let shared = batch_writer.shared_batches.builder( + desc.shared_id, + batch_writer.persist_client.clone(), + batch_writer.shard_id, + batch_writer.shard_name.clone(), + schemas.clone(), + desc.lower.clone(), + desc.upper.clone(), + ); + + let mut builder = PartBuilder::new(&*schemas.key, &*schemas.val); + while let Some(chunk) = updates_rx.recv().await { + for ((k, v), t, d) in &chunk { + builder.push(k, v, *t, *d); + if builder.len() >= SHARED_PART_LEN { + let part = builder.finish_and_replace(&*schemas.key, &*schemas.val); + shared.push(part).await; + } + } + } + let part = builder.finish(); + shared.push(part).await; + + let batch = shared.finish().await; + batch.map(|batch| batch.into_transmittable_batch()) + } + /// State maintained by the `write` operator on the Timely thread. struct State { sink_id: GlobalId, diff --git a/test/testdrive/mv-sink-shared-batches.td b/test/testdrive/mv-sink-shared-batches.td new file mode 100644 index 0000000000000..b2432ef966966 --- /dev/null +++ b/test/testdrive/mv-sink-shared-batches.td @@ -0,0 +1,81 @@ +# 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. + +# Coalesced ("shared") batch writing in the MV sink must produce results that +# are identical to the per-worker path. Exercise it across multiple workers in +# a single process, with churn (retractions and inserts), and assert that the +# materialized view always equals the equivalent one-shot aggregation. + +$ postgres-connect name=mz_system url=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr} + +$ postgres-execute connection=mz_system +ALTER SYSTEM SET enable_compute_sync_mv_sink_shared_batches = true + +# Multiple workers in one process, so the sink coalesces their parts. +> CREATE CLUSTER shared_batches SIZE 'scale=1,workers=4' + +> SET cluster = shared_batches + +> CREATE TABLE t (k bigint, v bigint) + +> INSERT INTO t SELECT x, x * 10 FROM generate_series(1, 20000) x + +> CREATE MATERIALIZED VIEW mv AS + SELECT k % 256 AS bucket, count(*) AS cnt, sum(v) AS s + FROM t + GROUP BY k % 256 + +> SELECT count(*) FROM mv +256 + +# The view must equal the one-shot aggregation. Both directions of the +# difference must be empty. +> (SELECT k % 256, count(*), sum(v) FROM t GROUP BY k % 256) + EXCEPT ALL + (SELECT bucket, cnt, s FROM mv) + +> (SELECT bucket, cnt, s FROM mv) + EXCEPT ALL + (SELECT k % 256, count(*), sum(v) FROM t GROUP BY k % 256) + +# Churn: retract half the rows and insert new keys. This spreads updates across +# all four workers within a single batch interval, which is what the shared +# batch coalesces. +> DELETE FROM t WHERE k % 2 = 0 + +> INSERT INTO t SELECT x, x * 10 FROM generate_series(20001, 35000) x + +> (SELECT k % 256, count(*), sum(v) FROM t GROUP BY k % 256) + EXCEPT ALL + (SELECT bucket, cnt, s FROM mv) + +> (SELECT bucket, cnt, s FROM mv) + EXCEPT ALL + (SELECT k % 256, count(*), sum(v) FROM t GROUP BY k % 256) + +# Odd k in 1..20000 (10000 rows) plus 20001..35000 (15000 rows). +> SELECT sum(cnt) FROM mv +25000 + +# A write with no net change must not corrupt the view. Workers that end up +# with no data for the interval must still finish their shared batch, since any +# one of them may be the last holder responsible for delivering the batch. +> DELETE FROM t WHERE false + +> (SELECT bucket, cnt, s FROM mv) + EXCEPT ALL + (SELECT k % 256, count(*), sum(v) FROM t GROUP BY k % 256) + +> DROP MATERIALIZED VIEW mv +> DROP TABLE t +> SET cluster = quickstart +> DROP CLUSTER shared_batches + +$ postgres-execute connection=mz_system +ALTER SYSTEM RESET enable_compute_sync_mv_sink_shared_batches From a5cf1626082b680f4859be7d83a18c9990a5c38c Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Mon, 13 Jul 2026 20:51:34 +0200 Subject: [PATCH 3/3] compute: barrier discipline for MV sink shared batches Builds on the shared-batches sink work (its PR): best-effort coalescing only merges workers whose builder handles overlap in time, so a process still writes ~1.4 parts per append. Add an opt-in barrier that makes the process-local workers agree on a single batch per interval. Each of the workers_per_process local workers reports exactly once per batch id: a write (SharedBatchBuilder::finish) or, when it supersedes the description without writing, a skip (SharedBatches::note_skip from the Timely thread). The shared batch is finished only after every participant reports, and delivered to the last worker that actually wrote, which always holds an output capability. The wait is unbounded on purpose. There is no cross-worker clock, so a timeout would have nothing to reason about. Completion rests on every process-local worker reporting once per batch id; in steady state every broadcast description is eventually written or superseded by every local worker, so the barrier converges, and teardown is handled by aborting the write tasks. The batch is finished only after all participants have pushed, so no push can race a finished builder. The barrier is per process: SharedBatches is process-local, so a multi-process replica coalesces to one batch per process, which the append operator already combines into a single compare_and_append. Gated behind enable_compute_sync_mv_sink_shared_batches_barrier, default off, and not randomized across the CI suite (the unbounded wait is too risky for that); covered by the shared-batches testdrive (single- and multi-process) and the parallel-workload flag flipper. Measured on an 8-worker replica: parts per append drop from ~1.4 to 1.0, cutting blob PUTs a further ~30%. Co-Authored-By: Claude Opus 4.8 (1M context) --- misc/python/materialize/mzcompose/__init__.py | 4 + .../materialize/parallel_workload/action.py | 3 + src/compute-types/src/dyncfgs.rs | 12 + src/compute/src/sink/materialized_view_v2.rs | 71 +++- src/storage-operators/src/persist.rs | 312 ++++++++++++++++-- test/testdrive/mv-sink-shared-batches.td | 48 ++- 6 files changed, 414 insertions(+), 36 deletions(-) diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 3dcc2f853679a..b5c07958614de 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -622,6 +622,10 @@ def get_default_system_parameters( # all. Only add it in UNINTERESTING_SYSTEM_PARAMETERS if none of the above # apply. UNINTERESTING_SYSTEM_PARAMETERS = [ + # The shared-batch barrier adds an unbounded cross-worker wait, too risky to randomize across + # the whole CI suite. It is covered by a dedicated testdrive test and the parallel-workload flag + # flipper. (The best-effort shared_batches flag stays in get_variable_system_parameters.) + "enable_compute_sync_mv_sink_shared_batches_barrier", "enable_compute_half_join2", "enable_mz_join_core", "linear_join_yielding", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 618aa4fe65914..e5ee71217e4b6 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -2885,6 +2885,9 @@ def __init__( self.flags_with_values["enable_compute_sync_mv_sink_shared_batches"] = ( BOOLEAN_FLAG_VALUES ) + self.flags_with_values["enable_compute_sync_mv_sink_shared_batches_barrier"] = ( + BOOLEAN_FLAG_VALUES + ) self.flags_with_values["enable_public_metrics_endpoint"] = BOOLEAN_FLAG_VALUES self.flags_with_values["persist_batch_structured_key_lower_len"] = [ "0", diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index fb891a3d1d687..0b7f03e80c923 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -272,6 +272,17 @@ pub const ENABLE_SYNC_MV_SINK_SHARED_BATCHES: Config = Config::new( ParameterScope::Environment, ); +/// When shared batches are enabled, make the workers in one process barrier on each batch interval +/// so they coalesce into exactly one batch, rather than best-effort. The barrier waits for every +/// process-local worker to report, so there is no timeout. +pub const ENABLE_SYNC_MV_SINK_SHARED_BATCHES_BARRIER: Config = Config::new( + "enable_compute_sync_mv_sink_shared_batches_barrier", + false, + "Barrier the workers in one process onto a single shared batch per interval in the MV sink, \ + rather than coalescing best-effort.", + ParameterScope::Environment, +); + /// Whether rendering should use the new MV sink correction buffer implementation. pub const ENABLE_CORRECTION_V2: Config = Config::new( "enable_compute_correction_v2", @@ -699,6 +710,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&ENABLE_MZ_JOIN_CORE) .add(&ENABLE_SYNC_MV_SINK) .add(&ENABLE_SYNC_MV_SINK_SHARED_BATCHES) + .add(&ENABLE_SYNC_MV_SINK_SHARED_BATCHES_BARRIER) .add(&ENABLE_CORRECTION_V2) .add(&CORRECTION_V2_CHAIN_PROPORTIONALITY) .add(&CORRECTION_V2_CHUNK_SIZE) diff --git a/src/compute/src/sink/materialized_view_v2.rs b/src/compute/src/sink/materialized_view_v2.rs index 7430f669f9169..406ddcdc1f709 100644 --- a/src/compute/src/sink/materialized_view_v2.rs +++ b/src/compute/src/sink/materialized_view_v2.rs @@ -53,7 +53,8 @@ use std::sync::Arc; use differential_dataflow::{Hashable, VecCollection}; use mz_compute_types::dyncfgs::{ - ENABLE_SYNC_MV_SINK_SHARED_BATCHES, MV_SINK_ADVANCE_PERSIST_FRONTIERS, + ENABLE_SYNC_MV_SINK_SHARED_BATCHES, ENABLE_SYNC_MV_SINK_SHARED_BATCHES_BARRIER, + MV_SINK_ADVANCE_PERSIST_FRONTIERS, }; use mz_dyncfg::ConfigSet; use mz_ore::cast::CastFrom; @@ -64,7 +65,7 @@ use mz_persist_types::ShardId; use mz_persist_types::codec_impls::UnitSchema; use mz_persist_types::part::PartBuilder; use mz_repr::{Diff, GlobalId, Row, Timestamp}; -use mz_storage_operators::persist::SharedBatches; +use mz_storage_operators::persist::{SharedBatchId, SharedBatches}; use mz_storage_types::StorageDiff; use mz_storage_types::sources::SourceData; use timely::PartialOrder; @@ -137,6 +138,7 @@ pub(super) fn persist_sink<'s>( descs.clone(), read_only_rx, Rc::clone(&compute_state.worker_config), + compute_state.workers_per_process, ); append::render(sink_id, persist_api, descs, batches); @@ -586,6 +588,7 @@ mod write { descs: DescsStream<'s>, mut read_only_rx: watch::Receiver, worker_config: Rc, + workers_per_process: usize, ) -> BatchesStream<'s> { let scope = desired.ok.scope(); let worker_id = scope.index(); @@ -663,9 +666,23 @@ mod write { let advance_persist_frontiers_at_startup = MV_SINK_ADVANCE_PERSIST_FRONTIERS.get(&worker_config); - // Read the shared-batch flag on the Timely thread. `worker_config` is an `Rc` - // and thus not `Send`, so the value must be captured before it moves into the Tokio task. + // Read the shared-batch flags on the Timely thread. `worker_config` is an `Rc` + // and thus not `Send`, so the values must be captured before it moves into the Tokio task. let shared_batches_enabled = ENABLE_SYNC_MV_SINK_SHARED_BATCHES.get(&worker_config); + // In barrier mode the process-local workers coalesce into exactly one batch per interval. + // `barrier` carries the local participant count; `None` keeps the best-effort discipline. + // + // INVARIANT: every process-local worker must render this sink in the same mode. A barrier + // worker waits (with no timeout) for a report from each of `workers_per_process` peers, so + // if one peer ran best-effort instead it would never report and the barrier would wedge + // permanently. This holds because dataflow render and config updates arrive as one ordered + // command stream, so all local workers read the same flag values at render time. + let barrier_enabled = ENABLE_SYNC_MV_SINK_SHARED_BATCHES_BARRIER.get(&worker_config); + let barrier = (shared_batches_enabled && barrier_enabled).then_some(workers_per_process); + // A worker that supersedes a description without writing it must release its barrier slot, + // so the shared batch does not wait for a batch this worker will never contribute to. + let skip_reporter = + barrier.map(|participants| (persist_api.persist_batches.clone(), participants)); // Mirror the persist-frontier initialization performed by `State::new` below. With the // flag enabled, `State` advances its Timely-side `persist_frontiers` to `as_of`, opening @@ -697,7 +714,8 @@ mod write { mz_ore::task::spawn( || operator_name(sink_id, "write::batch_writer"), async move { - let batch_writer = BatchWriter::new(&persist_api, shared_batches_enabled).await; + let batch_writer = + BatchWriter::new(&persist_api, shared_batches_enabled, barrier).await; while let Some(cmd) = cmd_rx.recv().await { corrections = @@ -740,6 +758,7 @@ mod write { as_of, advance_persist_frontiers_at_startup, read_only, + skip_reporter, ); // Whether a batch write is currently in flight in the Tokio task. @@ -874,11 +893,18 @@ mod write { schemas: Schemas, shared_batches: SharedBatches, shared_batches_enabled: bool, + /// `Some(participants)` selects the barrier coalescing discipline (see + /// [`SharedBatches::builder`]). + barrier: Option, writer: WriteHandle, } impl BatchWriter { - async fn new(persist_api: &PersistApi, shared_batches_enabled: bool) -> Self { + async fn new( + persist_api: &PersistApi, + shared_batches_enabled: bool, + barrier: Option, + ) -> Self { let writer = persist_api.open_writer().await; Self { persist_client: persist_api.open_client().await, @@ -895,6 +921,7 @@ mod write { }, shared_batches: persist_api.persist_batches.clone(), shared_batches_enabled, + barrier, writer, } } @@ -1069,10 +1096,10 @@ mod write { /// worker's parts with those of the other workers in the process that build the same interval. /// /// All workers building an interval share `desc.shared_id` (the `mint` operator broadcasts one - /// description), so their parts land in a single shared batch. Only the last worker to `finish` - /// receives that batch; the rest get `None`, which maps to the empty-batch response the write - /// operator already handles. Every worker must `finish` even when it pushed nothing, since any - /// one of them may be the last holder and thus responsible for delivering all workers' data. + /// description), so their parts land in a single shared batch. The handle that receives the + /// finished batch delivers it; the rest get `None`, which maps to the empty-batch response the + /// write operator already handles. Every worker must `finish` even when it pushed nothing, since + /// it may be the one that ends up delivering all workers' data. async fn write_shared_batch( batch_writer: &BatchWriter, desc: &BatchDescription, @@ -1087,6 +1114,7 @@ mod write { schemas.clone(), desc.lower.clone(), desc.upper.clone(), + batch_writer.barrier, ); let mut builder = PartBuilder::new(&*schemas.key, &*schemas.val); @@ -1135,6 +1163,9 @@ mod write { /// batches, so the `WriteBatch` path never sweeps `consolidate_before(upper)` forward; /// the forced consolidation stands in for it and is re-armed as long as this holds. read_only: bool, + /// In shared-batch barrier mode, the [`SharedBatches`] handle and process-local participant + /// count used to release a barrier slot when a description is superseded without a write. + skip_reporter: Option<(SharedBatches, usize)>, } impl State { @@ -1144,6 +1175,7 @@ mod write { as_of: Antichain, advance_persist_frontiers_at_startup: bool, read_only: bool, + skip_reporter: Option<(SharedBatches, usize)>, ) -> Self { // Force a consolidation of corrections after the snapshot updates have been fully // processed, to ensure we get rid of those as quickly as possible. @@ -1157,6 +1189,7 @@ mod write { batch_description: None, force_consolidation_after, read_only, + skip_reporter, }; // Immediately advance the persist frontier tracking to the `as_of`. @@ -1279,18 +1312,34 @@ mod write { // (invariant 1), so a regression means this description is outdated. We cannot use // `persist_frontiers` for the same check, because during snapshot processing those // frontiers can be ahead of the shard's write frontier and a still-valid description - // may have a `lower` below them. + // may have a `lower` before them. if let Some((prev, _)) = &self.batch_description { if PartialOrder::less_than(&desc.lower, &prev.lower) { self.trace(format!("skipping outdated batch description: {desc:?}")); + // This worker will not write `desc`, so release its barrier slot. + self.note_skip(desc.shared_id); return; } } + // Replacing an unwritten description means this worker will not write it (a written + // description is taken out of `batch_description` by `maybe_start_batch`). Release its + // barrier slot. + if let Some((prev, _)) = &self.batch_description { + self.note_skip(prev.shared_id); + } + self.batch_description = Some((desc, cap)); self.trace("set batch description"); } + /// In barrier mode, release the barrier slot for a description this worker will not write. + fn note_skip(&self, shared_id: SharedBatchId) { + if let Some((shared_batches, participants)) = &self.skip_reporter { + shared_batches.note_skip(shared_id, *participants); + } + } + /// Check if a batch can be written and send a write command to the Tokio task if so. fn maybe_start_batch( &mut self, diff --git a/src/storage-operators/src/persist.rs b/src/storage-operators/src/persist.rs index 3bedb13fec9ed..0efd3e2940ab5 100644 --- a/src/storage-operators/src/persist.rs +++ b/src/storage-operators/src/persist.rs @@ -18,8 +18,10 @@ use mz_storage_types::sources::SourceData; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; use std::fmt::Debug; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Weak}; use timely::progress::Antichain; +use tokio::sync::{Notify, oneshot}; use uuid::Uuid; /// A unique identifier for a [SharedBatchBuilder]. @@ -47,7 +49,18 @@ impl SharedBatchId { /// A struct from which to obtain a [SharedBatchBuilder]. #[derive(Debug, Clone)] pub struct SharedBatches { - data: Arc>)>>, + data: Arc>, + /// Hands out a unique id to each [SharedBatchBuilder], so the barrier can name a deliverer. + next_id: Arc, +} + +#[derive(Debug, Default)] +struct Inner { + last_retained_len: usize, + states: BTreeMap>, + /// Skips reported (see [SharedBatches::note_skip]) for a batch id before any worker in this + /// process opened a builder for it. Consumed when the builder is finally created. + pending_skips: BTreeMap, } fn upgrade_or_init(weak: &mut Weak, init: impl FnOnce() -> T) -> Arc { @@ -64,6 +77,7 @@ impl SharedBatches { pub fn new() -> Self { Self { data: Arc::default(), + next_id: Arc::new(AtomicU64::new(0)), } } @@ -72,14 +86,25 @@ impl SharedBatches { /// shared batch; if not, a new batch will be created. /// /// This is designed so that, in a multi-worker dataflow, workers that are building the same batch - /// at the same time can write fewer, larger parts instead of many small ones. It does not guarantee - /// that there will be _precisely_ one batch across all workers, however. + /// at the same time can write fewer, larger parts instead of many small ones. /// /// The batch's description is fixed by whichever caller creates it, so every caller sharing a /// `shared_batch_id` must pass the same `shard_id`, `lower`, and `upper`. Callers derive them /// from a single broadcast description, and the assertions below hold them to it: a caller /// whose description disagreed would otherwise have its updates silently written under someone /// else's bounds. + /// + /// `barrier` selects the coalescing discipline: + /// + /// * `None` (best-effort): the batch is finished by whichever handle calls + /// [SharedBatchBuilder::finish] last, determined by handle liveness. Workers whose handles do + /// not overlap in time land in separate batches, so this does not guarantee a single batch + /// across the process. + /// * `Some(participants)` (barrier): the batch is not finished until all `participants` + /// process-local workers have reported, either by finishing their handle or by + /// [Self::note_skip]. All parts then land in one batch, delivered to the last worker that + /// actually wrote. There is no timeout: completion relies on every participant reporting + /// exactly once (see [SharedBatchBuilder::finish]). pub fn builder( &self, shared_batch_id: SharedBatchId, @@ -89,21 +114,24 @@ impl SharedBatches { schemas: Schemas, lower: Antichain, upper: Antichain, + barrier: Option, ) -> SharedBatchBuilder { + let id = self.next_id.fetch_add(1, Ordering::Relaxed); let mut guard = self.data.lock().unwrap(); - let (last_retained_len, state_map) = &mut *guard; + let inner = &mut *guard; // We clean out entries from the map where all shared batch handles have been dropped. // Amortize by only scanning the map once it's doubled in size. - if state_map.len() > *last_retained_len * 2 { + if inner.states.len() > inner.last_retained_len * 2 { // Probe the count rather than `upgrade()`: the temporary strong reference an upgrade // creates can make a concurrent `SharedBatchBuilder::finish` see a strong count above // one and hand the batch to nobody, which seals the interval with no data. A stale // count is harmless in both directions, since zero cannot become live again and a // stale non-zero only keeps a dead entry until the next sweep. - state_map.retain(|_, weak| weak.strong_count() > 0); - *last_retained_len = state_map.len(); + inner.states.retain(|_, weak| weak.strong_count() > 0); + inner.last_retained_len = inner.states.len(); } - let weak = state_map.entry(shared_batch_id).or_default(); + let pending_skips = inner.pending_skips.remove(&shared_batch_id).unwrap_or(0); + let weak = inner.states.entry(shared_batch_id).or_default(); let desc = BatchDesc { shard_id, lower: lower.clone(), @@ -112,14 +140,23 @@ impl SharedBatches { let state = upgrade_or_init(weak, || { let (tx, mut rx) = tokio::sync::mpsc::channel(4); let task_id = format!("shared-batch-{shared_batch_id:?}-{shard_id}",); + // In barrier mode the initial `pending` counts every process-local participant, minus + // any that already reported a skip before this builder existed. + let pending = barrier.map(|participants| participants.saturating_sub(pending_skips)); BatchState { desc: desc.clone(), tx, + barrier: std::sync::Mutex::new(Barrier { + pending: pending.unwrap_or(0), + deliverer: None, + finalized: false, + }), + notify: Notify::new(), handle: mz_ore::task::spawn(|| task_id, async move { let mut builder = None; - while let Some(cmd) = rx.recv().await { - match cmd { - BatchCommand::Push(part) => { + loop { + match rx.recv().await { + Some(BatchCommand::Push(part)) => { let builder = builder.get_or_insert_with(|| { client.batch_builder( shard_id, @@ -131,13 +168,30 @@ impl SharedBatches { }); builder.add_part(part).await.expect("valid timestamps"); } + // Barrier mode: the deliverer explicitly finishes the batch and takes + // the result over the reply channel. + Some(BatchCommand::Finish(reply)) => { + let batch = match builder.take() { + Some(builder) => Some( + builder.finish(upper).await.expect("valid upper bound"), + ), + None => None, + }; + let _ = reply.send(batch); + return None; + } + // Best-effort mode: the channel closing (last handle dropped) is the + // signal to finish, and the result flows back through the join handle. + None => { + return match builder.take() { + Some(builder) => Some( + builder.finish(upper).await.expect("valid upper bound"), + ), + None => None, + }; + } } } - if let Some(builder) = builder { - Some(builder.finish(upper).await.expect("valid upper bound")) - } else { - None - } }), } }); @@ -148,9 +202,39 @@ impl SharedBatches { ); SharedBatchBuilder { + id, batch_id: shared_batch_id, shard_id, state, + barrier: barrier.is_some(), + } + } + + /// Report that a process-local worker will not contribute to the batch with the given id + /// (because it superseded the batch's description without writing it). Only meaningful in + /// barrier mode: it releases the barrier slot the worker would otherwise have filled by + /// finishing a builder, so a shared batch does not wait for a worker that will never write. + pub fn note_skip(&self, shared_batch_id: SharedBatchId, participants: usize) { + let mut guard = self.data.lock().unwrap(); + let inner = &mut *guard; + match inner.states.get(&shared_batch_id).and_then(Weak::upgrade) { + Some(state) => { + let mut barrier = state.barrier.lock().unwrap(); + barrier.arrive(); + if barrier.finalized { + state.notify.notify_waiters(); + } + } + None => { + // The builder for this id does not exist yet. Remember the skip so the barrier + // starts with a correspondingly smaller count once a worker opens the builder. + let count = inner.pending_skips.entry(shared_batch_id).or_default(); + *count += 1; + // If every participant skipped, no batch will ever be built for this id. + if *count >= participants { + inner.pending_skips.remove(&shared_batch_id); + } + } } } } @@ -165,29 +249,74 @@ struct BatchDesc { enum BatchCommand { Push(Part), + Finish(oneshot::Sender>>), } +/// Barrier coordinating the process-local workers that share a batch. #[derive(Debug)] +struct Barrier { + /// Participants that have not yet reported (by finishing a builder or skipping). + pending: usize, + /// The handle chosen to finish and deliver the batch: the last worker to actually write. + deliverer: Option, + finalized: bool, +} + +impl Barrier { + /// Record that a participant reported. Finalizes once the last one does. + fn arrive(&mut self) { + // Every process-local participant reports exactly once per batch id (a write or a skip). + // A report past zero means that invariant broke: the barrier would finalize early and a + // later push could race the finished builder. Fail loudly in test rather than mask it with + // the saturating decrement below. The decrement stays outside the assert so it is not + // compiled out in release builds. + debug_assert!( + self.pending > 0, + "shared batch barrier reported more times than participants" + ); + self.pending = self.pending.saturating_sub(1); + if self.pending == 0 { + self.finalized = true; + } + } +} + struct BatchState { desc: BatchDesc, tx: tokio::sync::mpsc::Sender, + barrier: std::sync::Mutex, + notify: Notify, handle: JoinHandle>>, } +impl Debug for BatchState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("BatchState") + .field("barrier", &self.barrier) + .finish_non_exhaustive() + } +} + /// A handle for a shared batch builder. Everyone with a handle for the same builder can -/// [Self::push] to that builder, but the last handle to call [Self::finish] will receive a batch -/// containing all the data. +/// [Self::push] to that builder. /// /// Note that it's quite important to call [Self::finish], even if you haven't pushed anything /// into the batch, in case you're holding the last builder for a shared batch. /// +/// In best-effort mode the last handle to call [Self::finish] receives the batch, so every handle +/// must call finish even if it pushed nothing. In barrier mode the batch is delivered to the last +/// handle that pushed data; the rest receive `None`. +/// /// Dropping the last handle instead still completes the batch, but hands it to nobody, so its /// parts stay in blob until persist's leaked-blob cleanup collects them. That is the teardown /// path, not a path any caller should take deliberately. pub struct SharedBatchBuilder { + id: u64, batch_id: SharedBatchId, shard_id: ShardId, state: Arc, + /// Whether this handle uses the barrier discipline rather than best-effort. + barrier: bool, } impl Debug for SharedBatchBuilder { @@ -215,15 +344,67 @@ impl SharedBatchBuilder { /// Fetch the results of the batch builder from the shared state, if any. /// - /// Only the last builder to call this method will obtain the resulting batch, and that batch - /// will include all data written by all workers. - /// This means that, for any batch interval that we actually want to append, all workers - /// must call finish even if they did not push any batches themselves. + /// See [SharedBatches::builder] for which handle receives the batch in each mode. pub async fn finish(self) -> Option> { + if self.barrier { + self.finish_barrier().await + } else { + self.finish_best_effort().await + } + } + + /// Best-effort finish: the last live handle drops the channel, which finishes the batch. + async fn finish_best_effort(self) -> Option> { let state = Arc::into_inner(self.state)?; drop(state.tx); // The task only finishes the batch once the channel is closed. state.handle.await } + + /// Barrier finish: report as a writer, wait for the other process-local participants, and if + /// this handle is the chosen deliverer, finish the batch and return it. + /// + /// The wait is unbounded on purpose. There is no cross-worker clock to justify a timeout, so + /// completion relies on the invariant that every process-local participant reports exactly once + /// per batch id, either here or via [SharedBatches::note_skip]. In steady state every broadcast + /// description is eventually written or superseded by every local worker, so the barrier + /// converges; on teardown the write tasks are aborted, which drops these futures. This means + /// the batch is only ever finished after all participants have pushed, so no push can race a + /// finished builder. + async fn finish_barrier(self) -> Option> { + // Register as the (current) deliverer candidate and arrive at the barrier. The last writer + // to arrive stays the candidate, and it always holds an output capability, so it can emit. + let done = { + let mut barrier = self.state.barrier.lock().unwrap(); + barrier.deliverer = Some(self.id); + barrier.arrive(); + barrier.finalized + }; + if done { + self.state.notify.notify_waiters(); + } else { + loop { + // Register for the wakeup before re-checking, so a notify between the check and the + // await is not lost. + let notified = self.state.notify.notified(); + if self.state.barrier.lock().unwrap().finalized { + break; + } + notified.await; + } + } + + let am_deliverer = self.state.barrier.lock().unwrap().deliverer == Some(self.id); + if !am_deliverer { + return None; + } + let (reply_tx, reply_rx) = oneshot::channel(); + self.state + .tx + .send(BatchCommand::Finish(reply_tx)) + .await + .expect("task failed"); + reply_rx.await.expect("task failed") + } } #[cfg(test)] @@ -273,6 +454,7 @@ mod tests { schemas.clone(), lower.clone(), upper.clone(), + None, ); let second = shared.builder( batch_id, @@ -282,6 +464,7 @@ mod tests { schemas.clone(), lower.clone(), upper.clone(), + None, ); let part = bool_part(&schemas); first.push(part.clone()).await; @@ -302,4 +485,89 @@ mod tests { "batch should include updates from both pushes" ) } + + // In barrier mode the batch waits for every participant, then delivers all pushes to exactly + // one handle, regardless of finish order. + #[mz_ore::test(tokio::test)] + async fn test_shared_batch_barrier() { + let client = PersistClient::new_for_tests().await; + let shared = SharedBatches::new(); + let batch_id = SharedBatchId::new(); + let shard_id = ShardId::new(); + let schemas = bool_schemas(); + let lower = Antichain::from_elem(0.into()); + let upper = Antichain::from_elem(1.into()); + let barrier = Some(2); + + let make = |id| { + shared.builder( + id, + client.clone(), + shard_id, + "test".to_string(), + schemas.clone(), + lower.clone(), + upper.clone(), + barrier, + ) + }; + let first = make(batch_id); + let second = make(batch_id); + first.push(bool_part(&schemas)).await; + second.push(bool_part(&schemas)).await; + + // Finish concurrently: the barrier must resolve without either call being awaited first. + let (a, b) = tokio::join!(first.finish(), second.finish()); + assert!( + a.is_some() ^ b.is_some(), + "exactly one handle should receive the batch" + ); + let batch = a.or(b).unwrap(); + assert_eq!(batch.shard_id(), shard_id); + assert_eq!( + batch.into_hollow_batch().len, + 2, + "the shared batch must contain both workers' pushes" + ); + } + + // A skip releases a participant's barrier slot, so the batch finishes without waiting for a + // worker that will never write. + #[mz_ore::test(tokio::test)] + async fn test_shared_batch_barrier_skip() { + let client = PersistClient::new_for_tests().await; + let shared = SharedBatches::new(); + let batch_id = SharedBatchId::new(); + let shard_id = ShardId::new(); + let schemas = bool_schemas(); + let lower = Antichain::from_elem(0.into()); + let upper = Antichain::from_elem(1.into()); + // Two participants: without the skip releasing the second slot, the sole writer's barrier + // would never complete and this test would hang. + let barrier = Some(2); + + let writer = shared.builder( + batch_id, + client.clone(), + shard_id, + "test".to_string(), + schemas.clone(), + lower.clone(), + upper.clone(), + barrier, + ); + writer.push(bool_part(&schemas)).await; + // The second participant skips instead of writing. + shared.note_skip(batch_id, 2); + + let batch = writer + .finish() + .await + .expect("sole writer delivers the batch"); + assert_eq!( + batch.into_hollow_batch().len, + 1, + "the batch contains the writer's push" + ); + } } diff --git a/test/testdrive/mv-sink-shared-batches.td b/test/testdrive/mv-sink-shared-batches.td index b2432ef966966..3cccaade6f291 100644 --- a/test/testdrive/mv-sink-shared-batches.td +++ b/test/testdrive/mv-sink-shared-batches.td @@ -8,14 +8,19 @@ # by the Apache License, Version 2.0. # Coalesced ("shared") batch writing in the MV sink must produce results that -# are identical to the per-worker path. Exercise it across multiple workers in -# a single process, with churn (retractions and inserts), and assert that the -# materialized view always equals the equivalent one-shot aggregation. +# are identical to the per-worker path. Exercise it across multiple workers, with +# churn (retractions and inserts), and assert that the materialized view always +# equals the equivalent one-shot aggregation. The barrier coalesces per process, +# so this covers both a single-process replica and a multi-process one, where the +# per-process batches are combined by the append operator. $ postgres-connect name=mz_system url=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr} $ postgres-execute connection=mz_system ALTER SYSTEM SET enable_compute_sync_mv_sink_shared_batches = true +# The barrier makes the workers coalesce into exactly one batch per interval, which exercises the +# barrier and skip-reporting paths. +ALTER SYSTEM SET enable_compute_sync_mv_sink_shared_batches_barrier = true # Multiple workers in one process, so the sink coalesces their parts. > CREATE CLUSTER shared_batches SIZE 'scale=1,workers=4' @@ -77,5 +82,42 @@ ALTER SYSTEM SET enable_compute_sync_mv_sink_shared_batches = true > SET cluster = quickstart > DROP CLUSTER shared_batches +# Multi-process replica (scale > 1): each process barriers its own workers into one +# batch, and the append operator combines the per-process batches into a single +# append. The result must still equal the one-shot aggregation. +> CREATE CLUSTER shared_batches_mp SIZE 'scale=2,workers=2' + +> SET cluster = shared_batches_mp + +> CREATE TABLE t (k bigint, v bigint) + +> INSERT INTO t SELECT x, x * 10 FROM generate_series(1, 20000) x + +> CREATE MATERIALIZED VIEW mv AS + SELECT k % 256 AS bucket, count(*) AS cnt, sum(v) AS s + FROM t + GROUP BY k % 256 + +> DELETE FROM t WHERE k % 2 = 0 + +> INSERT INTO t SELECT x, x * 10 FROM generate_series(20001, 35000) x + +> (SELECT k % 256, count(*), sum(v) FROM t GROUP BY k % 256) + EXCEPT ALL + (SELECT bucket, cnt, s FROM mv) + +> (SELECT bucket, cnt, s FROM mv) + EXCEPT ALL + (SELECT k % 256, count(*), sum(v) FROM t GROUP BY k % 256) + +> SELECT sum(cnt) FROM mv +25000 + +> DROP MATERIALIZED VIEW mv +> DROP TABLE t +> SET cluster = quickstart +> DROP CLUSTER shared_batches_mp + $ postgres-execute connection=mz_system ALTER SYSTEM RESET enable_compute_sync_mv_sink_shared_batches +ALTER SYSTEM RESET enable_compute_sync_mv_sink_shared_batches_barrier