From 9d14786c45f8df5fd9c6880d2193ae8832ed3628 Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Sat, 4 Jul 2026 07:40:09 +0000 Subject: [PATCH 1/2] storage-controller: respond to table appends before txn-wal apply The commit to the txns shard already makes a table write durable and linearized, so send the append response then and run the idempotent apply and compaction afterwards, off the caller's critical path. Reads of the just-written shards block until the apply happens, and a crash before applying is covered by the next append's apply_le. Co-Authored-By: Claude Fable 5 --- .../scenarios/benchmark_main.py | 69 +++++++++++++++++++ src/storage-controller/src/persist_handles.rs | 29 +++++--- 2 files changed, 88 insertions(+), 10 deletions(-) diff --git a/misc/python/materialize/feature_benchmark/scenarios/benchmark_main.py b/misc/python/materialize/feature_benchmark/scenarios/benchmark_main.py index 84b70745e0dc2..59e034198f2d9 100644 --- a/misc/python/materialize/feature_benchmark/scenarios/benchmark_main.py +++ b/misc/python/materialize/feature_benchmark/scenarios/benchmark_main.py @@ -2716,6 +2716,75 @@ def benchmark(self) -> MeasurementSource: """) +class MultiTableTransactionCommit(Coordinator): + """Measure COMMIT latency of a write transaction spanning many tables. + + Such a transaction goes through group commit as one txn-wal transaction + touching one data shard per table, so COMMIT pays the txns-shard write + plus whatever per-shard work sits on the group commit response path + (batch apply and compaction, unless deferred). + + The table write worker is a single serial task: it will not start the + next txn's txns-shard write until the previous txn's apply and compaction + have finished. So to observe the response-path latency of one COMMIT in + isolation, the worker must be idle when that COMMIT arrives. The reads at + the top of each iteration drain it: they touch every table, which blocks + until the previous iteration's txn is fully applied across all shards. + Reading a single table would only wait for that one shard's apply and + leave the worker busy with the rest, so the measured COMMIT would queue + behind that leftover work and hide any response-path win. + """ + + SCALE = 3 # 1000 tables + # The deferred apply/compaction the response path saves is one persist + # write per shard, so the win grows with the table count while the fixed + # per-COMMIT floor (group commit, timestamp selection, pgwire) does not. + # A high table count is what makes that saving dominate the measured + # window rather than the floor. + FIXED_SCALE = True + + def init(self) -> list[Action]: + creates = "\n".join( + f"> CREATE TABLE mttc_t{i} (x INT);" for i in range(self.n()) + ) + return [TdAction(f""" +$ postgres-execute connection=postgres://mz_system:materialize@${{testdrive.materialize-internal-sql-addr}} +ALTER SYSTEM SET max_tables = {self.n() + 200}; + +{creates} +""")] + + def benchmark(self) -> MeasurementSource: + inserts = "\n".join( + f"> INSERT INTO mttc_t{i} VALUES (1);" for i in range(self.n()) + ) + # Read every table so this waits for the whole previous txn to apply, + # draining the serial write worker before we measure. Separate reads + # (rather than one wide UNION ALL) keep each peek cheap to plan and + # avoid the deep-expr recursion limit at high table counts. + drain = "\n".join( + f"> SELECT count(*) >= 0 FROM mttc_t{i}\ntrue" for i in range(self.n()) + ) + + return Td(f""" +{drain} + +> BEGIN + +{inserts} + +> SELECT 1; + /* A */ +1 + +> COMMIT + +> SELECT 1; + /* B */ +1 +""") + + class ReplicaExpiration(Scenario): # Causes "tried to kill container, but did not receive an exit event" errors when killing container afterwards SCALE = 5 diff --git a/src/storage-controller/src/persist_handles.rs b/src/storage-controller/src/persist_handles.rs index 2d901f247f9ff..b8dfcb2ee2b45 100644 --- a/src/storage-controller/src/persist_handles.rs +++ b/src/storage-controller/src/persist_handles.rs @@ -476,10 +476,21 @@ impl TxnsTableWorker { // Sneak in any txns shard tidying from previous commits. txn.tidy(std::mem::take(&mut self.tidy)); let txn_res = txn.commit_at(&mut self.txns, write_ts).await; - let response = match txn_res { + match txn_res { Ok(apply) => { - // TODO: Do the applying in a background task. This will be a - // significant INSERT latency performance win. + // The commit to the txns shard already made the write durable + // and linearized, so unblock the caller before applying. The + // apply is idempotent and reads of the affected data shards + // block until it has happened, so deferring it shifts latency + // from every write onto reads of the just-written shards. If + // we crash before applying, the next append's apply covers it, + // because applying is `apply_le`: it applies all unapplied + // txns up to its timestamp, and group commit appends at least + // once per timestamp interval. + // + // It is not an error for the other end to hang up. + let _ = tx.send(Ok(())); + debug!("applying {:?}", apply); let tidy = apply.apply(&mut self.txns).await; self.tidy.merge(tidy); @@ -488,8 +499,6 @@ impl TxnsTableWorker { // and compact as aggressively as we can (i.e. to the time we // just wrote). let () = self.txns.compact_to(write_ts).await; - - Ok(()) } Err(current) => { self.tidy.merge(txn.take_tidy()); @@ -497,7 +506,7 @@ impl TxnsTableWorker { "unable to commit txn at {:?} current={:?}", write_ts, current ); - Err(StorageError::InvalidUppers( + let response = Err(StorageError::InvalidUppers( self.write_handles .keys() .copied() @@ -506,11 +515,11 @@ impl TxnsTableWorker { current_upper: Antichain::from_elem(current), }) .collect(), - )) + )); + // It is not an error for the other end to hang up. + let _ = tx.send(response); } - }; - // It is not an error for the other end to hang up. - let _ = tx.send(response); + } } } From ccb9bde8cf1624fce57dae6e19b2503b32fa7a4e Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Fri, 10 Jul 2026 07:39:58 +0000 Subject: [PATCH 2/2] Address reviwer comments --- doc/user/data/metrics.yml | 14 ++++++++++++++ src/storage-client/src/metrics.rs | 17 ++++++++++++++++- src/storage-controller/src/lib.rs | 6 +++--- src/storage-controller/src/persist_handles.rs | 13 ++++++++++++- 4 files changed, 45 insertions(+), 5 deletions(-) diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index 9d3f435d20fd5..4765e8714fb1e 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -3478,6 +3478,20 @@ metrics: - replica_id source: src/storage-client/src/metrics.rs visibility: internal +- name: mz_storage_controller_table_apply_seconds_bucket + help: Latency of applying a durable table append to its data shards, performed off the write critical path after the append response is returned. + labels: + - le + source: src/storage-client/src/metrics.rs + visibility: internal +- name: mz_storage_controller_table_apply_seconds_count + help: Latency of applying a durable table append to its data shards, performed off the write critical path after the append response is returned. + source: src/storage-client/src/metrics.rs + visibility: internal +- name: mz_storage_controller_table_apply_seconds_sum + help: Latency of applying a durable table append to its data shards, performed off the write critical path after the append response is returned. + source: src/storage-client/src/metrics.rs + visibility: internal - name: mz_storage_regressed_offset_known help: number of regressed offset_known stats for this id labels: diff --git a/src/storage-client/src/metrics.rs b/src/storage-client/src/metrics.rs index af996ec1234d9..890fe776fe4c4 100644 --- a/src/storage-client/src/metrics.rs +++ b/src/storage-client/src/metrics.rs @@ -17,9 +17,10 @@ use mz_cluster_client::metrics::{ControllerMetrics, WallclockLagMetrics}; use mz_ore::cast::CastFrom; use mz_ore::metric; use mz_ore::metrics::{ - CounterVec, DeleteOnDropCounter, DeleteOnDropGauge, IntCounterVec, MetricsRegistry, + CounterVec, DeleteOnDropCounter, DeleteOnDropGauge, Histogram, IntCounterVec, MetricsRegistry, UIntGaugeVec, }; +use mz_ore::stats::histogram_seconds_buckets; use mz_repr::GlobalId; use mz_service::transport; use mz_storage_types::instances::StorageInstanceId; @@ -47,6 +48,9 @@ pub struct StorageControllerMetrics { replica_connects_total: IntCounterVec, replica_connect_wait_time_seconds_total: CounterVec, + // table writes + table_apply_seconds: Histogram, + /// Metrics shared with the compute controller. shared: ControllerMetrics, } @@ -99,11 +103,22 @@ impl StorageControllerMetrics { help: "The total time the storage controller spent waiting for replica (re-)connection.", var_labels: ["instance_id", "replica_id"], )), + table_apply_seconds: metrics_registry.register(metric!( + name: "mz_storage_controller_table_apply_seconds", + help: "Latency of applying a durable table append to its data shards, performed off the write critical path after the append response is returned.", + buckets: histogram_seconds_buckets(0.128, 32.0), + )), shared, } } + /// Returns the histogram tracking the latency of applying a durable table + /// append to its data shards. + pub fn table_apply_seconds(&self) -> Histogram { + self.table_apply_seconds.clone() + } + pub fn regressed_offset_known( &self, id: mz_repr::GlobalId, diff --git a/src/storage-controller/src/lib.rs b/src/storage-controller/src/lib.rs index 60fd4da7aff32..ad9e13e9bb9e5 100644 --- a/src/storage-controller/src/lib.rs +++ b/src/storage-controller/src/lib.rs @@ -2738,6 +2738,8 @@ where .get_txn_wal_shard() .expect("must call prepare initialization before creating storage controller"); + let metrics = StorageControllerMetrics::new(metrics_registry, controller_metrics); + let persist_table_worker = if read_only { let txns_write = txns_client .open_writer( @@ -2763,7 +2765,7 @@ where ) .await; txns.upgrade_version().await; - persist_handles::PersistTableWriteWorker::new_txns(txns) + persist_handles::PersistTableWriteWorker::new_txns(txns, metrics.table_apply_seconds()) }; let txns_read = TxnsRead::start::(txns_client.clone(), txns_id).await; @@ -2783,8 +2785,6 @@ where let (instance_response_tx, instance_response_rx) = mpsc::unbounded_channel(); - let metrics = StorageControllerMetrics::new(metrics_registry, controller_metrics); - let now_dt = mz_ore::now::to_datetime(now()); Self { diff --git a/src/storage-controller/src/persist_handles.rs b/src/storage-controller/src/persist_handles.rs index b8dfcb2ee2b45..7fab1fc0a07da 100644 --- a/src/storage-controller/src/persist_handles.rs +++ b/src/storage-controller/src/persist_handles.rs @@ -19,6 +19,7 @@ use futures::future::BoxFuture; use futures::stream::FuturesUnordered; use futures::{FutureExt, StreamExt}; use itertools::Itertools; +use mz_ore::metrics::{Histogram, MetricsFutureExt}; use mz_ore::tracing::OpenTelemetryContext; use mz_persist_client::ShardId; use mz_persist_client::write::WriteHandle; @@ -178,6 +179,7 @@ impl PersistTableWriteWorker { pub(crate) fn new_txns( txns: TxnsHandle, + apply_seconds: Histogram, ) -> Self { let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<(tracing::Span, PersistTableWriteCmd)>(); @@ -186,6 +188,7 @@ impl PersistTableWriteWorker { txns, write_handles: BTreeMap::new(), tidy: Tidy::default(), + apply_seconds, }; worker.run(rx).await }); @@ -281,6 +284,10 @@ struct TxnsTableWorker { txns: TxnsHandle, write_handles: BTreeMap, tidy: Tidy, + /// Latency of the deferred apply, which runs after the append response is + /// returned. This is the coverage `mz_append_table_duration_seconds` loses + /// by responding before applying. + apply_seconds: Histogram, } impl TxnsTableWorker { @@ -492,7 +499,11 @@ impl TxnsTableWorker { let _ = tx.send(Ok(())); debug!("applying {:?}", apply); - let tidy = apply.apply(&mut self.txns).await; + let tidy = apply + .apply(&mut self.txns) + .wall_time() + .observe(self.apply_seconds.clone()) + .await; self.tidy.merge(tidy); // We don't serve any reads out of this TxnsHandle, so go ahead