From d3289ea79d3de143e64a236472dd50a8d235a114 Mon Sep 17 00:00:00 2001 From: Qindeel Ishtiaq Date: Mon, 6 Jul 2026 13:51:03 +0200 Subject: [PATCH 1/3] adapter: move arrangement size snapshotting off the coordinator loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mz_object_arrangement_size_history snapshot did its persist reads, consolidation, and row building inline on the coordinator main loop. This was blocking the main coordinator work, so the arrangement_size_history_collection_interval dyncfg is currently set to disabled in staging and production. Restructure the collection to mirror the storage_usage_fetch / storage_usage_update split: * ArrangementSizesSnapshot now only resolves catalog ids on the main loop, then spawns a task that takes a read timestamp from a shared oracle handle, snapshots mz_object_arrangement_sizes and mz_compute_hydration_times, consolidates, and prepares records. * A new ArrangementSizesWrite message carries the prepared records back to the coordinator, which stamps the collection timestamp, packs rows, and hands the append to an async task. The read timestamp is taken inside the spawned task immediately before the reads, so the window for the collections' since to overtake it is smaller than before. A failed snapshot skips the cycle and reschedules, as before. Measured locally at a 10s collection interval: the snapshot handler's main-loop time drops from ~163ms (the full collection time, by construction) to ~61µs over 31 cycles, while the collection work runs on the spawned task. Co-Authored-By: Claude Fable 5 --- src/adapter/src/coord.rs | 13 + src/adapter/src/coord/message_handler.rs | 423 ++++++++++++++--------- 2 files changed, 280 insertions(+), 156 deletions(-) diff --git a/src/adapter/src/coord.rs b/src/adapter/src/coord.rs index 371309530596d..6203842dd1f9b 100644 --- a/src/adapter/src/coord.rs +++ b/src/adapter/src/coord.rs @@ -316,6 +316,17 @@ impl IdPool { } } +/// A row for `mz_object_arrangement_size_history`, prepared off-thread by the +/// arrangement sizes snapshot task and stamped with a collection timestamp at +/// write time. +#[derive(Debug)] +pub struct ArrangementSizeRecord { + pub replica_id: String, + pub object_id: String, + pub size: i64, + pub hydration_complete: bool, +} + #[derive(Debug)] pub enum Message { Command(OpenTelemetryContext, Command), @@ -359,6 +370,7 @@ pub enum Message { StorageUsagePrune(Vec), ArrangementSizesSchedule, ArrangementSizesSnapshot, + ArrangementSizesWrite(Vec), ArrangementSizesPrune(Vec), /// Performs any cleanup and logging actions necessary for /// finalizing a statement execution. @@ -509,6 +521,7 @@ impl Message { Message::StorageUsagePrune(_) => "storage_usage_prune", Message::ArrangementSizesSchedule => "arrangement_sizes_schedule", Message::ArrangementSizesSnapshot => "arrangement_sizes_snapshot", + Message::ArrangementSizesWrite(_) => "arrangement_sizes_write", Message::ArrangementSizesPrune(_) => "arrangement_sizes_prune", Message::RetireExecute { .. } => "retire_execute", Message::ExecuteSingleStatementTransaction { .. } => { diff --git a/src/adapter/src/coord/message_handler.rs b/src/adapter/src/coord/message_handler.rs index 017096579b00e..4602f6a8ea771 100644 --- a/src/adapter/src/coord/message_handler.rs +++ b/src/adapter/src/coord/message_handler.rs @@ -32,6 +32,7 @@ use mz_sql::ast::Statement; use mz_sql::names::ResolvedIds; use mz_sql::pure::PurifiedStatement; use mz_storage_client::controller::IntrospectionType; +use mz_storage_types::StorageDiff; use opentelemetry::trace::TraceContextExt; use rand::{Rng, SeedableRng, rngs}; use serde_json::json; @@ -42,7 +43,7 @@ use crate::active_compute_sink::{ActiveComputeSink, ActiveComputeSinkRetireReaso use crate::catalog::BuiltinTableUpdate; use crate::command::Command; use crate::coord::{ - AlterConnectionValidationReady, ClusterReplicaStatuses, Coordinator, + AlterConnectionValidationReady, ArrangementSizeRecord, ClusterReplicaStatuses, Coordinator, CreateConnectionValidationReady, Message, PurifiedStatementReady, WatchSetResponse, }; use crate::telemetry::{EventDetails, SegmentClientExt}; @@ -142,6 +143,9 @@ impl Coordinator { Message::ArrangementSizesSnapshot => { self.arrangement_sizes_snapshot().boxed_local().await; } + Message::ArrangementSizesWrite(records) => { + self.arrangement_sizes_write(records).boxed_local().await; + } Message::ArrangementSizesPrune(expired) => { self.arrangement_sizes_prune(expired).boxed_local().await; } @@ -457,28 +461,25 @@ impl Coordinator { }); } - /// Snapshots the current contents of `mz_object_arrangement_sizes` and - /// appends them to `mz_object_arrangement_size_history`, tagged with a - /// shared `collection_timestamp`. Reschedules on completion. + /// Kicks off a snapshot of `mz_object_arrangement_sizes` for appending to + /// `mz_object_arrangement_size_history`. /// - /// Each `(replica_id, object_id)` pair is recorded with a - /// `hydration_complete` flag derived from `mz_compute_hydration_times`: - /// `true` once the pair's initial hydration on that replica is finished, - /// `false` while still building. Consumers that want only stable sizes - /// should filter `WHERE hydration_complete`. + /// The persist reads and row preparation are too slow for the coordinator + /// main loop, so they run on a spawned task. The prepared records come + /// back as [`Message::ArrangementSizesWrite`] and are appended by + /// [`Coordinator::arrangement_sizes_write`], which also reschedules the + /// next collection. An empty or failed snapshot reschedules directly. #[mz_ore::instrument(level = "debug")] - async fn arrangement_sizes_snapshot(&mut self) { - // The catalog server is not writable in read-only mode. + async fn arrangement_sizes_snapshot(&self) { + // The catalog server is not writable in read-only mode. Skip the + // cycle and reschedule so collection resumes once the coordinator + // transitions out of read-only. The transition is one-way, so + // `arrangement_sizes_write` needs no check of its own. if self.controller.read_only() { self.schedule_arrangement_sizes_collection().await; return; } - let collection_timer = self - .metrics - .arrangement_sizes_collection_time_seconds - .start_timer(); - let live_item_id = self.catalog().resolve_builtin_storage_collection( &mz_catalog::builtin::MZ_OBJECT_ARRANGEMENT_SIZES_UNIFIED, ); @@ -490,67 +491,63 @@ impl Coordinator { .catalog .get_entry(&hydration_item_id) .latest_global_id(); - let history_item_id = self - .catalog() - .resolve_builtin_table(&mz_catalog::builtin::MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY); - let read_ts = self.get_local_read_ts().await; - let snapshot = match self - .controller - .storage_collections - .snapshot(live_global_id, read_ts) - .await - { - Ok(s) => s, - Err(e) => { - tracing::warn!("arrangement sizes snapshot failed: {e:?}"); - drop(collection_timer); - self.schedule_arrangement_sizes_collection().await; - return; - } - }; - let mut hydration_snapshot = match self - .controller - .storage_collections - .snapshot(hydration_global_id, read_ts) - .await - { - Ok(s) => s, - Err(e) => { - tracing::warn!("arrangement sizes hydration snapshot failed: {e:?}"); - drop(collection_timer); - self.schedule_arrangement_sizes_collection().await; - return; - } - }; - differential_dataflow::consolidation::consolidate(&mut hydration_snapshot); - - // Build the set of pairs whose initial hydration has finished - // (`time_ns IS NOT NULL`). The set drives the `hydration_complete` - // flag for each row we emit below. - let mut datum_vec = mz_repr::DatumVec::new(); - let mut hydrated: BTreeSet<(String, String)> = BTreeSet::new(); - const HYDRATION_COL_REPLICA_ID: usize = 0; - const HYDRATION_COL_OBJECT_ID: usize = 1; - const HYDRATION_COL_TIME_NS: usize = 2; - const HYDRATION_COL_COUNT: usize = 3; - for (row, diff) in &hydration_snapshot { - if *diff != 1 { - continue; - } - let datums = datum_vec.borrow_with(row); - if datums.len() < HYDRATION_COL_COUNT { - continue; - } - if datums[HYDRATION_COL_TIME_NS].is_null() { - continue; + let oracle = self.get_local_timestamp_oracle(); + let storage_collections = Arc::clone(&self.controller.storage_collections); + let collection_metric = self + .metrics + .arrangement_sizes_collection_time_seconds + .clone(); + let internal_cmd_tx = self.internal_cmd_tx.clone(); + + task::spawn(|| "arrangement_sizes_snapshot", async move { + let collection_metric_timer = collection_metric.start_timer(); + + // Taking `read_ts` inside the task keeps the window between + // choosing the timestamp and reading minimal. If the collection's + // since still overtakes it, the cycle is skipped and the + // reschedule retries at the next interval. + let read_ts = oracle.read_ts().await; + let live_snapshot = match storage_collections.snapshot(live_global_id, read_ts).await { + Ok(s) => s, + Err(e) => { + tracing::warn!("arrangement sizes snapshot failed: {e:?}"); + let _ = internal_cmd_tx.send(Message::ArrangementSizesSchedule); + return; + } + }; + let hydration_snapshot = match storage_collections + .snapshot(hydration_global_id, read_ts) + .await + { + Ok(s) => s, + Err(e) => { + tracing::warn!("arrangement sizes hydration snapshot failed: {e:?}"); + let _ = internal_cmd_tx.send(Message::ArrangementSizesSchedule); + return; + } + }; + + let records = arrangement_sizes_records(live_snapshot, hydration_snapshot); + collection_metric_timer.observe_duration(); + + let msg = if records.is_empty() { + Message::ArrangementSizesSchedule + } else { + Message::ArrangementSizesWrite(records) + }; + // It is not an error for this task to outlive `internal_cmd_rx`. + if let Err(e) = internal_cmd_tx.send(msg) { + warn!("internal_cmd_rx dropped before we could send: {e:?}"); } - hydrated.insert(( - datums[HYDRATION_COL_REPLICA_ID].unwrap_str().to_string(), - datums[HYDRATION_COL_OBJECT_ID].unwrap_str().to_string(), - )); - } + }); + } + /// Stamps prepared snapshot records with a shared `collection_timestamp` + /// and appends them to `mz_object_arrangement_size_history`. Reschedules + /// the next collection once the append completes. + #[mz_ore::instrument(level = "debug")] + async fn arrangement_sizes_write(&mut self, records: Vec) { // `collection_ts` is stamped after the snapshot so it's always >= the // state the rows describe, and monotone across restarts. The snapshot // read and this stamp aren't atomic, but the resulting skew is bounded @@ -562,90 +559,43 @@ impl Coordinator { .expect("collection_timestamp must fit into TimestampTz"), ); - let mut consolidated = snapshot; - differential_dataflow::consolidation::consolidate(&mut consolidated); - - // Column positions in `mz_object_arrangement_sizes`. - const LIVE_COL_REPLICA_ID: usize = 0; - const LIVE_COL_OBJECT_ID: usize = 1; - const LIVE_COL_SIZE: usize = 2; - const LIVE_COL_COUNT: usize = 3; - - let mut skipped_malformed: u64 = 0; - let mut skipped_null_size: u64 = 0; - let mut updates: Vec = Vec::with_capacity(consolidated.len()); - for (row, diff) in consolidated.iter() { - if *diff != 1 { - continue; - } - let datums = datum_vec.borrow_with(row); - // Surface schema drift via a warn log below rather than silently - // skipping entire snapshots. - if datums.len() != LIVE_COL_COUNT { - skipped_malformed += 1; - continue; - } - let replica_id = datums[LIVE_COL_REPLICA_ID].unwrap_str(); - let object_id = datums[LIVE_COL_OBJECT_ID].unwrap_str(); - let size_datum = datums[LIVE_COL_SIZE]; - // The history table's `size` is non-null; fabricating zero would - // be misleading, so drop. - if size_datum.is_null() { - skipped_null_size += 1; - continue; - } - let size = size_datum.unwrap_int64(); - // Pairs whose hydration hasn't completed yet are still recorded, - // tagged with `hydration_complete = false`. Consumers that care - // only about stable sizes can filter on `hydration_complete`. - let hydration_complete = - hydrated.contains(&(replica_id.to_string(), object_id.to_string())); - let new_row = Row::pack_slice(&[ - Datum::String(replica_id), - Datum::String(object_id), - Datum::Int64(size), - collection_datum, - Datum::from(hydration_complete), - ]); - updates.push(BuiltinTableUpdate::row(history_item_id, new_row, Diff::ONE)); - } - if skipped_malformed > 0 { - warn!( - "mz_object_arrangement_sizes schema drift: skipped {skipped_malformed} rows \ - with unexpected arity" - ); - } - if skipped_null_size > 0 { - tracing::debug!("skipped {skipped_null_size} live rows with null size"); - } + let history_item_id = self + .catalog() + .resolve_builtin_table(&mz_catalog::builtin::MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY); + + let updates: Vec<_> = records + .into_iter() + .map(|record| { + let row = Row::pack_slice(&[ + Datum::String(&record.replica_id), + Datum::String(&record.object_id), + Datum::Int64(record.size), + collection_datum, + Datum::from(record.hydration_complete), + ]); + BuiltinTableUpdate::row(history_item_id, row, Diff::ONE) + }) + .collect(); let row_count = updates.len(); - // Captures snapshot + row construction. The async table-apply below - // is captured separately by `mz_append_table_duration_seconds`. - collection_timer.observe_duration(); - - if !updates.is_empty() { - self.metrics - .arrangement_sizes_rows_written - .inc_by(u64::cast_from(row_count)); - // TODO(arrangement-sizes): when the writeable-catalog-server plumbing - // in https://github.com/MaterializeInc/materialize/pull/35436 lands, - // append directly on `mz_catalog_server` instead of going through - // the environmentd builtin-table-update path. - let (fut, _) = self.builtin_table_update().execute(updates).await; - let internal_cmd_tx = self.internal_cmd_tx.clone(); - let task_span = - info_span!(parent: None, "coord::arrangement_sizes_snapshot::table_updates"); - OpenTelemetryContext::obtain().attach_as_parent_to(&task_span); - task::spawn(|| "arrangement_sizes_snapshot_apply", async move { - fut.instrument(task_span).await; - if let Err(e) = internal_cmd_tx.send(Message::ArrangementSizesSchedule) { - warn!("internal_cmd_rx dropped before we could send: {e:?}"); - } - }); - } else { - self.schedule_arrangement_sizes_collection().await; - } + self.metrics + .arrangement_sizes_rows_written + .inc_by(u64::cast_from(row_count)); + + // TODO(arrangement-sizes): when the writeable-catalog-server plumbing + // in https://github.com/MaterializeInc/materialize/pull/35436 lands, + // append directly on `mz_catalog_server` instead of going through + // the environmentd builtin-table-update path. + let (fut, _) = self.builtin_table_update().execute(updates).await; + let internal_cmd_tx = self.internal_cmd_tx.clone(); + let task_span = info_span!(parent: None, "coord::arrangement_sizes_write::table_updates"); + OpenTelemetryContext::obtain().attach_as_parent_to(&task_span); + task::spawn(|| "arrangement_sizes_write_table_updates", async move { + fut.instrument(task_span).await; + if let Err(e) = internal_cmd_tx.send(Message::ArrangementSizesSchedule) { + warn!("internal_cmd_rx dropped before we could send: {e:?}"); + } + }); tracing::debug!( "appended {row_count} rows to mz_object_arrangement_size_history at ts {collection_ts}" @@ -1169,3 +1119,164 @@ impl Coordinator { } } } + +/// Builds history records from snapshots of `mz_object_arrangement_sizes` and +/// `mz_compute_hydration_times`. +/// +/// Each `(replica_id, object_id)` pair is recorded with a +/// `hydration_complete` flag: `true` once the pair's initial hydration on that +/// replica is finished (`time_ns IS NOT NULL`), `false` while still building. +/// Consumers that want only stable sizes should filter +/// `WHERE hydration_complete`. +fn arrangement_sizes_records( + mut live_snapshot: Vec<(Row, StorageDiff)>, + mut hydration_snapshot: Vec<(Row, StorageDiff)>, +) -> Vec { + differential_dataflow::consolidation::consolidate(&mut live_snapshot); + differential_dataflow::consolidation::consolidate(&mut hydration_snapshot); + + let mut datum_vec = mz_repr::DatumVec::new(); + + // Column positions in `mz_compute_hydration_times`. + const HYDRATION_COL_REPLICA_ID: usize = 0; + const HYDRATION_COL_OBJECT_ID: usize = 1; + const HYDRATION_COL_TIME_NS: usize = 2; + const HYDRATION_COL_COUNT: usize = 3; + + let mut hydrated: BTreeSet<(String, String)> = BTreeSet::new(); + for (row, diff) in &hydration_snapshot { + if *diff != 1 { + continue; + } + let datums = datum_vec.borrow_with(row); + if datums.len() < HYDRATION_COL_COUNT { + continue; + } + if datums[HYDRATION_COL_TIME_NS].is_null() { + continue; + } + hydrated.insert(( + datums[HYDRATION_COL_REPLICA_ID].unwrap_str().to_string(), + datums[HYDRATION_COL_OBJECT_ID].unwrap_str().to_string(), + )); + } + + // Column positions in `mz_object_arrangement_sizes`. + const LIVE_COL_REPLICA_ID: usize = 0; + const LIVE_COL_OBJECT_ID: usize = 1; + const LIVE_COL_SIZE: usize = 2; + const LIVE_COL_COUNT: usize = 3; + + let mut skipped_malformed: u64 = 0; + let mut skipped_null_size: u64 = 0; + let mut records = Vec::with_capacity(live_snapshot.len()); + for (row, diff) in &live_snapshot { + if *diff != 1 { + continue; + } + let datums = datum_vec.borrow_with(row); + // Surface schema drift via a warn log below rather than silently + // skipping entire snapshots. + if datums.len() != LIVE_COL_COUNT { + skipped_malformed += 1; + continue; + } + let replica_id = datums[LIVE_COL_REPLICA_ID].unwrap_str(); + let object_id = datums[LIVE_COL_OBJECT_ID].unwrap_str(); + let size_datum = datums[LIVE_COL_SIZE]; + // The history table's `size` is non-null; fabricating zero would + // be misleading, so drop. + if size_datum.is_null() { + skipped_null_size += 1; + continue; + } + let hydration_complete = + hydrated.contains(&(replica_id.to_string(), object_id.to_string())); + records.push(ArrangementSizeRecord { + replica_id: replica_id.to_string(), + object_id: object_id.to_string(), + size: size_datum.unwrap_int64(), + hydration_complete, + }); + } + if skipped_malformed > 0 { + warn!( + "mz_object_arrangement_sizes schema drift: skipped {skipped_malformed} rows \ + with unexpected arity" + ); + } + if skipped_null_size > 0 { + tracing::debug!("skipped {skipped_null_size} live rows with null size"); + } + records +} + +#[cfg(test)] +mod arrangement_sizes_records_tests { + use mz_repr::{Datum, Row}; + + use super::arrangement_sizes_records; + + fn live_row(replica_id: &str, object_id: &str, size: Option) -> Row { + Row::pack_slice(&[ + Datum::String(replica_id), + Datum::String(object_id), + size.map_or(Datum::Null, Datum::Int64), + ]) + } + + fn hydration_row(replica_id: &str, object_id: &str, hydrated: bool) -> Row { + Row::pack_slice(&[ + Datum::String(replica_id), + Datum::String(object_id), + if hydrated { + Datum::Int64(1) + } else { + Datum::Null + }, + ]) + } + + #[mz_ore::test] + fn hydration_flag_per_pair() { + let live = vec![ + (live_row("u1", "u100", Some(10)), 1), + (live_row("u1", "u200", Some(20)), 1), + ]; + let hydration = vec![ + (hydration_row("u1", "u100", true), 1), + (hydration_row("u1", "u200", false), 1), + ]; + let records = arrangement_sizes_records(live, hydration); + assert_eq!(records.len(), 2); + assert!( + records + .iter() + .any(|r| r.object_id == "u100" && r.hydration_complete) + ); + assert!( + records + .iter() + .any(|r| r.object_id == "u200" && !r.hydration_complete) + ); + } + + #[mz_ore::test] + fn skips_malformed_null_and_retracted() { + let live = vec![ + // Wrong arity. + (Row::pack_slice(&[Datum::String("u1")]), 1), + // Null size. + (live_row("u1", "u100", None), 1), + // Retracted by consolidation. + (live_row("u1", "u200", Some(20)), 1), + (live_row("u1", "u200", Some(20)), -1), + (live_row("u1", "u300", Some(30)), 1), + ]; + let records = arrangement_sizes_records(live, Vec::new()); + assert_eq!(records.len(), 1); + assert_eq!(records[0].object_id, "u300"); + assert_eq!(records[0].size, 30); + assert!(!records[0].hydration_complete); + } +} From a74c28f162de7d53e1bbe254a9ccd093be031488 Mon Sep 17 00:00:00 2001 From: Qindeel Ishtiaq Date: Mon, 6 Jul 2026 14:05:27 +0200 Subject: [PATCH 2/3] adapter: don't record stale arrangement sizes after a restart The collections backing the arrangement size history snapshots, mz_object_arrangement_sizes and mz_compute_hydration_times, retain rows from before an environmentd restart until the new introspection subscribes deliver their first data and the shards are reconciled. Track on each introspection subscribe when it first appended data in this process, and only record history rows for replicas whose sizes and hydration subscribes first delivered at least 10s ago. The margin covers the collection manager's batched flush and the oracle read timestamp trailing the wall clock. Without it, a snapshot racing the first append could still read the pre-restart shard contents. Stale hydration data is likewise no longer trusted: rows only get hydration_complete = true if the replica's hydration subscribe is fresh. Adds a restart workflow that kills and restarts environmentd five times at a 500ms collection interval and asserts that no post-restart history row carries a pre-restart size. Fails without this fix with 299 stale rows in the first round. Fixes SQL-218. Co-Authored-By: Claude Fable 5 --- src/adapter/src/coord/introspection.rs | 38 +++++++++ src/adapter/src/coord/message_handler.rs | 95 ++++++++++++++++++++- test/restart/mzcompose.py | 104 +++++++++++++++++++++++ 3 files changed, 233 insertions(+), 4 deletions(-) diff --git a/src/adapter/src/coord/introspection.rs b/src/adapter/src/coord/introspection.rs index dc6743613332e..de5d579c626d6 100644 --- a/src/adapter/src/coord/introspection.rs +++ b/src/adapter/src/coord/introspection.rs @@ -28,6 +28,9 @@ //! failure), `handle_introspection_subscribe_batch` reacts on the corresponding error responses //! by reinstalling the failed introspection subscribes. +use std::collections::BTreeSet; +use std::time::{Duration, Instant}; + use anyhow::bail; use derivative::Derivative; use mz_adapter_types::dyncfgs::ENABLE_INTROSPECTION_SUBSCRIBES; @@ -72,6 +75,13 @@ pub(super) struct IntrospectionSubscribe { /// introspection data around in the meantime makes for a better UX than removing it. #[derivative(Debug = "ignore")] deferred_write: Option, + /// When this subscribe first appended data to the target storage collection, if it has. + /// + /// Until then, the target collection may still contain rows written by a previous incarnation + /// of this subscribe (before an environmentd restart, or before the target replica + /// reconnected). Consumers that must not observe such stale rows, like the + /// `mz_object_arrangement_size_history` snapshots, use this to judge per-replica freshness. + first_data_at: Option, } impl IntrospectionSubscribe { @@ -144,6 +154,7 @@ impl Coordinator { replica_id, spec, deferred_write: None, + first_data_at: None, }; self.introspection_subscribes.insert(id, subscribe); @@ -410,6 +421,9 @@ impl Coordinator { // Ensure that the contents of the target storage collection are cleaned when the new // subscribe starts reporting data. subscribe.deferred_write = Some(subscribe.delete_write_op()); + // Until then, the collection serves the previous subscribe's data, which the replica may + // have invalidated by restarting. + subscribe.first_data_at = None; self.introspection_subscribes.insert(new_id, subscribe); self.sequence_introspection_subscribe(new_id, spec, cluster_id, replica_id) @@ -468,6 +482,8 @@ impl Coordinator { .update_introspection_collection(subscribe.spec.introspection_type, op); } + subscribe.first_data_at.get_or_insert_with(Instant::now); + self.controller.storage.update_introspection_collection( subscribe.spec.introspection_type, StorageWriteOp::Append { @@ -475,6 +491,28 @@ impl Coordinator { }, ); } + + /// Returns the IDs of replicas whose introspection subscribe of the given type first + /// delivered data at least `margin` ago. + /// + /// Rows for other replicas in the corresponding storage collection are not trustworthy: they + /// either predate this environmentd process or were written before the replica reconnected, + /// and may describe a previous incarnation of the replica. The margin accounts for the + /// subscribe's first append becoming visible to readers only asynchronously (the collection + /// manager flushes writes in batches, and snapshot reads use an oracle timestamp that trails + /// the wall clock). + pub(super) fn fresh_introspection_replicas( + &self, + introspection_type: IntrospectionType, + margin: Duration, + ) -> BTreeSet { + self.introspection_subscribes + .values() + .filter(|s| s.spec.introspection_type == introspection_type) + .filter(|s| s.first_data_at.is_some_and(|at| at.elapsed() >= margin)) + .map(|s| s.replica_id.to_string()) + .collect() + } } impl Staged for IntrospectionSubscribeStage { diff --git a/src/adapter/src/coord/message_handler.rs b/src/adapter/src/coord/message_handler.rs index 4602f6a8ea771..08420ddaa50a0 100644 --- a/src/adapter/src/coord/message_handler.rs +++ b/src/adapter/src/coord/message_handler.rs @@ -469,6 +469,10 @@ impl Coordinator { /// back as [`Message::ArrangementSizesWrite`] and are appended by /// [`Coordinator::arrangement_sizes_write`], which also reschedules the /// next collection. An empty or failed snapshot reschedules directly. + /// + /// Rows from replicas without fresh introspection data are excluded, so + /// sizes predating an environmentd or replica restart are not recorded. + /// See [`Coordinator::fresh_introspection_replicas`]. #[mz_ore::instrument(level = "debug")] async fn arrangement_sizes_snapshot(&self) { // The catalog server is not writable in read-only mode. Skip the @@ -480,6 +484,25 @@ impl Coordinator { return; } + // See `fresh_introspection_replicas` for why the margin is needed. + // 10s comfortably covers the collection manager's ~1s write batching + // plus the oracle read timestamp trailing the wall clock. + const FRESHNESS_MARGIN: Duration = Duration::from_secs(10); + let fresh_size_replicas = self.fresh_introspection_replicas( + IntrospectionType::ComputeObjectArrangementSizes, + FRESHNESS_MARGIN, + ); + let fresh_hydration_replicas = self.fresh_introspection_replicas( + IntrospectionType::ComputeHydrationTimes, + FRESHNESS_MARGIN, + ); + if fresh_size_replicas.is_empty() { + // No replica has reported sizes in this process yet, so the live + // collection contains only stale rows (or none). Skip the cycle. + self.schedule_arrangement_sizes_collection().await; + return; + } + let live_item_id = self.catalog().resolve_builtin_storage_collection( &mz_catalog::builtin::MZ_OBJECT_ARRANGEMENT_SIZES_UNIFIED, ); @@ -528,7 +551,12 @@ impl Coordinator { } }; - let records = arrangement_sizes_records(live_snapshot, hydration_snapshot); + let records = arrangement_sizes_records( + live_snapshot, + hydration_snapshot, + &fresh_size_replicas, + &fresh_hydration_replicas, + ); collection_metric_timer.observe_duration(); let msg = if records.is_empty() { @@ -1128,9 +1156,15 @@ impl Coordinator { /// replica is finished (`time_ns IS NOT NULL`), `false` while still building. /// Consumers that want only stable sizes should filter /// `WHERE hydration_complete`. +/// +/// Rows from replicas outside `fresh_size_replicas` are dropped, and the +/// hydration flag is only trusted for replicas in `fresh_hydration_replicas`. +/// Rows for other replicas may predate an environmentd or replica restart. fn arrangement_sizes_records( mut live_snapshot: Vec<(Row, StorageDiff)>, mut hydration_snapshot: Vec<(Row, StorageDiff)>, + fresh_size_replicas: &BTreeSet, + fresh_hydration_replicas: &BTreeSet, ) -> Vec { differential_dataflow::consolidation::consolidate(&mut live_snapshot); differential_dataflow::consolidation::consolidate(&mut hydration_snapshot); @@ -1155,8 +1189,12 @@ fn arrangement_sizes_records( if datums[HYDRATION_COL_TIME_NS].is_null() { continue; } + let replica_id = datums[HYDRATION_COL_REPLICA_ID].unwrap_str(); + if !fresh_hydration_replicas.contains(replica_id) { + continue; + } hydrated.insert(( - datums[HYDRATION_COL_REPLICA_ID].unwrap_str().to_string(), + replica_id.to_string(), datums[HYDRATION_COL_OBJECT_ID].unwrap_str().to_string(), )); } @@ -1169,6 +1207,7 @@ fn arrangement_sizes_records( let mut skipped_malformed: u64 = 0; let mut skipped_null_size: u64 = 0; + let mut skipped_stale_replica: u64 = 0; let mut records = Vec::with_capacity(live_snapshot.len()); for (row, diff) in &live_snapshot { if *diff != 1 { @@ -1182,6 +1221,10 @@ fn arrangement_sizes_records( continue; } let replica_id = datums[LIVE_COL_REPLICA_ID].unwrap_str(); + if !fresh_size_replicas.contains(replica_id) { + skipped_stale_replica += 1; + continue; + } let object_id = datums[LIVE_COL_OBJECT_ID].unwrap_str(); let size_datum = datums[LIVE_COL_SIZE]; // The history table's `size` is non-null; fabricating zero would @@ -1208,11 +1251,19 @@ fn arrangement_sizes_records( if skipped_null_size > 0 { tracing::debug!("skipped {skipped_null_size} live rows with null size"); } + if skipped_stale_replica > 0 { + tracing::debug!( + "skipped {skipped_stale_replica} live rows from replicas without fresh \ + introspection data" + ); + } records } #[cfg(test)] mod arrangement_sizes_records_tests { + use std::collections::BTreeSet; + use mz_repr::{Datum, Row}; use super::arrangement_sizes_records; @@ -1237,6 +1288,10 @@ mod arrangement_sizes_records_tests { ]) } + fn replicas(ids: &[&str]) -> BTreeSet { + ids.iter().map(|id| id.to_string()).collect() + } + #[mz_ore::test] fn hydration_flag_per_pair() { let live = vec![ @@ -1247,7 +1302,8 @@ mod arrangement_sizes_records_tests { (hydration_row("u1", "u100", true), 1), (hydration_row("u1", "u200", false), 1), ]; - let records = arrangement_sizes_records(live, hydration); + let fresh = replicas(&["u1"]); + let records = arrangement_sizes_records(live, hydration, &fresh, &fresh); assert_eq!(records.len(), 2); assert!( records @@ -1273,10 +1329,41 @@ mod arrangement_sizes_records_tests { (live_row("u1", "u200", Some(20)), -1), (live_row("u1", "u300", Some(30)), 1), ]; - let records = arrangement_sizes_records(live, Vec::new()); + let fresh = replicas(&["u1"]); + let records = arrangement_sizes_records(live, Vec::new(), &fresh, &fresh); assert_eq!(records.len(), 1); assert_eq!(records[0].object_id, "u300"); assert_eq!(records[0].size, 30); assert!(!records[0].hydration_complete); } + + #[mz_ore::test] + fn skips_rows_from_stale_replicas() { + // u1 has fresh introspection data, u2's rows predate a restart. + let live = vec![ + (live_row("u1", "u100", Some(10)), 1), + (live_row("u2", "u100", Some(99)), 1), + ]; + let hydration = vec![ + (hydration_row("u1", "u100", true), 1), + (hydration_row("u2", "u100", true), 1), + ]; + let fresh = replicas(&["u1"]); + let records = arrangement_sizes_records(live, hydration, &fresh, &fresh); + assert_eq!(records.len(), 1); + assert_eq!(records[0].replica_id, "u1"); + assert!(records[0].hydration_complete); + } + + #[mz_ore::test] + fn stale_hydration_data_is_not_trusted() { + // u1's sizes subscribe is fresh but its hydration subscribe is not, + // so its stale "hydrated" row must not mark the record complete. + let live = vec![(live_row("u1", "u100", Some(10)), 1)]; + let hydration = vec![(hydration_row("u1", "u100", true), 1)]; + let records = + arrangement_sizes_records(live, hydration, &replicas(&["u1"]), &replicas(&[])); + assert_eq!(records.len(), 1); + assert!(!records[0].hydration_complete); + } } diff --git a/test/restart/mzcompose.py b/test/restart/mzcompose.py index ae51a6b55bc20..f69e383cd155a 100644 --- a/test/restart/mzcompose.py +++ b/test/restart/mzcompose.py @@ -1081,6 +1081,110 @@ def workflow_rename_schema_types_functions(c: Composition) -> None: c.sql("DROP SCHEMA s2") +def workflow_arrangement_sizes_stale_snapshot_after_restart(c: Composition) -> None: + """After a restart, mz_object_arrangement_size_history should not + contain rows with outdated sizes tagged hydration_complete = true. + + The collections backing the history snapshots retain pre-restart rows + until the new introspection subscribes replace them, so snapshots taken + in that window would record outdated sizes (SQL-218). + + Inserting data right after each restart makes the post-restart + steady-state sizes larger than the pre-restart sizes. Any + hydration_complete = true row still carrying the old size was + captured from the stale persist shard before the new introspection + subscribe replaced it. + """ + + num_replicas = 2 + names = tuple(f"sidx{i}" for i in range(1, 21)) + expected_count = len(names) * num_replicas + name_filter = "(" + ", ".join(f"'{n}'" for n in names) + ")" + + c.down(destroy_volumes=True) + with c.override( + Materialized( + additional_system_parameter_defaults={ + "arrangement_size_history_collection_interval": "500ms", + }, + sanity_restart=False, + ) + ): + c.up("materialized") + c.sql(dedent(f"""\ + CREATE CLUSTER stale_test SIZE 'scale=1,workers=1', REPLICATION FACTOR {num_replicas}; + CREATE TABLE stale_t (a int, b text); + INSERT INTO stale_t SELECT g, repeat('x', 1024) FROM generate_series(1, 30000) g; + CREATE VIEW stale_v AS SELECT a, b FROM stale_t; + {"".join(f"CREATE INDEX sidx{i} IN CLUSTER stale_test ON stale_v ((a + {i}));" for i in range(1, 21))} + """)) + + def wait_for_full_sample() -> None: + deadline = time.time() + 120 + while time.time() < deadline: + if c.sql_query(f""" + SELECT 1 FROM mz_internal.mz_object_arrangement_size_history h + JOIN mz_objects o ON o.id = h.object_id + WHERE o.name IN {name_filter} + GROUP BY h.collection_timestamp + HAVING count(*) = {expected_count} LIMIT 1"""): + return + time.sleep(0.5) + raise UIError("timed out waiting for a full sample") + + def get_live_sizes() -> dict[tuple[str, str], int]: + return {(r, n): int(sz) for r, n, sz in c.sql_query(f""" + SELECT s.replica_id, o.name, s.size + FROM mz_internal.mz_object_arrangement_sizes s + JOIN mz_objects o ON o.id = s.object_id + WHERE o.name IN {name_filter}""")} + + wait_for_full_sample() + + for round_num in range(5): + pre_sizes = get_live_sizes() + max_ts = c.sql_query(f""" + SELECT max(h.collection_timestamp)::text + FROM mz_internal.mz_object_arrangement_size_history h + JOIN mz_objects o ON o.id = h.object_id + WHERE o.name IN {name_filter}""")[0][0] + + c.kill("materialized") + c.up("materialized") + + lo = 30001 + round_num * 30000 + c.sql( + f"INSERT INTO stale_t SELECT g, repeat('x', 1024) FROM generate_series({lo}, {lo + 29999}) g" + ) + time.sleep(8) + + wait_for_full_sample() + post_sizes = get_live_sizes() + assert len(post_sizes) == expected_count + + mismatches = [ + (ts, rid, n, int(sz), int(post_sizes[(rid, n)])) + for ts, rid, n, sz in c.sql_query(f""" + SELECT h.collection_timestamp::text, h.replica_id, o.name, h.size + FROM mz_internal.mz_object_arrangement_size_history h + JOIN mz_objects o ON o.id = h.object_id + WHERE o.name IN {name_filter} + AND h.hydration_complete + AND h.collection_timestamp > '{max_ts}'::timestamptz + ORDER BY h.collection_timestamp""") + if (rid, n) in post_sizes + and (rid, n) in pre_sizes + and int(sz) == pre_sizes[(rid, n)] + and int(sz) != post_sizes[(rid, n)] + ] + + assert not mismatches, ( + f"round {round_num}: {len(mismatches)} history rows with " + f"hydration_complete = true carry stale pre-restart sizes; " + f"first 10: {mismatches[:10]}" + ) + + def workflow_default(c: Composition) -> None: def process(name: str) -> None: if name == "default": From 52f8f8ffa0bcbfe5a09f9905cfdc101820ffa714 Mon Sep 17 00:00:00 2001 From: Qindeel Ishtiaq Date: Mon, 6 Jul 2026 14:06:03 +0200 Subject: [PATCH 3/3] adapter: report sub-10 MiB arrangements in mz_object_arrangement_sizes The subscribe feeding mz_object_arrangement_sizes filtered out objects whose arrangements totaled less than 10 MiB, so small objects never appeared in the collection at all and consumers could not distinguish "has small arrangements" from "has no arrangements". The column comment meanwhile promised exact sizes below 10 MiB, which the quantization never delivered. Drop the HAVING floor and keep the round-to-nearest-10-MiB quantization: arrangements below 5 MiB now report a size of 0 and larger ones their quantized size, with churn behavior unchanged. The column comment now describes the actual behavior. The history snapshot skips size-0 rows. The floor had been doubling as history volume control, and recording every small object on every cycle would bloat mz_object_arrangement_size_history without adding signal. Fixes SQL-271. Co-Authored-By: Claude Fable 5 --- src/adapter/src/coord/introspection.rs | 11 ++++---- src/adapter/src/coord/message_handler.rs | 28 +++++++++++++++++++ src/catalog/src/builtin/mz_internal.rs | 7 +++-- test/testdrive/arrangement-sizes.td | 34 ++++++++++++++++++++++-- 4 files changed, 68 insertions(+), 12 deletions(-) diff --git a/src/adapter/src/coord/introspection.rs b/src/adapter/src/coord/introspection.rs index de5d579c626d6..9a078c5c15dc4 100644 --- a/src/adapter/src/coord/introspection.rs +++ b/src/adapter/src/coord/introspection.rs @@ -614,11 +614,11 @@ const SUBSCRIBES: &[SubscribeSpec] = &[ // differential logs where each `+1` row represents one byte of heap delta; // after consolidation, `COUNT(*)` is the current arrangement size in bytes. // - // The `HAVING` floor drops objects below 10 MiB. Below that threshold the - // heap-size collection wiggles by a few bytes per second from ordinary - // allocator activity, and emitting exact bytes would push a downstream - // update on every wiggle. Quantizing to the nearest 10 MiB keeps the - // emitted size stable across in-bucket wiggle. + // Sizes are quantized to the nearest 10 MiB: the heap-size collection + // wiggles by a few bytes per second from ordinary allocator activity, and + // emitting exact bytes would push a downstream update on every wiggle. + // Arrangements below 5 MiB quantize to a size of 0. They are kept rather + // than dropped so that every arranged object is present in the collection. // // `mz_dataflow_addresses.address[1]` is the root of each operator's address // tree, which equals the owning `dataflow_id` — so we can go addresses → @@ -648,7 +648,6 @@ const SUBSCRIBES: &[SubscribeSpec] = &[ ) AS rs ON rs.operator_id = od.operator_id WHERE ce.export_id NOT LIKE 't%' GROUP BY ce.export_id - HAVING COUNT(*) >= 10485760 )", }, ]; diff --git a/src/adapter/src/coord/message_handler.rs b/src/adapter/src/coord/message_handler.rs index 08420ddaa50a0..b3bf08dcff5f0 100644 --- a/src/adapter/src/coord/message_handler.rs +++ b/src/adapter/src/coord/message_handler.rs @@ -1160,6 +1160,9 @@ impl Coordinator { /// Rows from replicas outside `fresh_size_replicas` are dropped, and the /// hydration flag is only trusted for replicas in `fresh_hydration_replicas`. /// Rows for other replicas may predate an environmentd or replica restart. +/// +/// Rows with a size of 0 (arrangements below the live collection's 5 MiB +/// quantization threshold) are not recorded. fn arrangement_sizes_records( mut live_snapshot: Vec<(Row, StorageDiff)>, mut hydration_snapshot: Vec<(Row, StorageDiff)>, @@ -1207,6 +1210,7 @@ fn arrangement_sizes_records( let mut skipped_malformed: u64 = 0; let mut skipped_null_size: u64 = 0; + let mut skipped_zero_size: u64 = 0; let mut skipped_stale_replica: u64 = 0; let mut records = Vec::with_capacity(live_snapshot.len()); for (row, diff) in &live_snapshot { @@ -1233,6 +1237,13 @@ fn arrangement_sizes_records( skipped_null_size += 1; continue; } + // A quantized size of 0 means "below 5 MiB". The live collection + // keeps such rows so small objects stay visible, but recording them + // every cycle would bloat the history with rows carrying no signal. + if size_datum.unwrap_int64() == 0 { + skipped_zero_size += 1; + continue; + } let hydration_complete = hydrated.contains(&(replica_id.to_string(), object_id.to_string())); records.push(ArrangementSizeRecord { @@ -1251,6 +1262,9 @@ fn arrangement_sizes_records( if skipped_null_size > 0 { tracing::debug!("skipped {skipped_null_size} live rows with null size"); } + if skipped_zero_size > 0 { + tracing::debug!("skipped {skipped_zero_size} live rows with zero size"); + } if skipped_stale_replica > 0 { tracing::debug!( "skipped {skipped_stale_replica} live rows from replicas without fresh \ @@ -1337,6 +1351,20 @@ mod arrangement_sizes_records_tests { assert!(!records[0].hydration_complete); } + #[mz_ore::test] + fn skips_zero_size_rows() { + // Size 0 means "below the live collection's quantization threshold". + // Such objects stay visible live but are not recorded in the history. + let live = vec![ + (live_row("u1", "u100", Some(0)), 1), + (live_row("u1", "u200", Some(10485760)), 1), + ]; + let fresh = replicas(&["u1"]); + let records = arrangement_sizes_records(live, Vec::new(), &fresh, &fresh); + assert_eq!(records.len(), 1); + assert_eq!(records[0].object_id, "u200"); + } + #[mz_ore::test] fn skips_rows_from_stale_replicas() { // u1 has fresh introspection data, u2's rows predate a restart. diff --git a/src/catalog/src/builtin/mz_internal.rs b/src/catalog/src/builtin/mz_internal.rs index 19c498e47414c..da1896112fe3b 100644 --- a/src/catalog/src/builtin/mz_internal.rs +++ b/src/catalog/src/builtin/mz_internal.rs @@ -4394,10 +4394,9 @@ pub static MZ_OBJECT_ARRANGEMENT_SIZES_UNIFIED: LazyLock = LazyLo ), ( "size", - "The total arrangement heap and batcher size in bytes for this object on this replica. \ - Objects smaller than 10 MiB are reported at their exact size; objects 10 MiB or larger \ - are rounded to the nearest 10 MiB boundary to reduce per-byte churn in the differential \ - collection.", + "The total arrangement heap and batcher size in bytes for this object on this replica, \ + rounded to the nearest 10 MiB boundary to reduce per-byte churn in the differential \ + collection. Objects with less than 5 MiB of arrangements report a size of 0.", ), ]), is_retained_metrics_object: true, diff --git a/test/testdrive/arrangement-sizes.td b/test/testdrive/arrangement-sizes.td index fca835fe318ff..333e2a310acd6 100644 --- a/test/testdrive/arrangement-sizes.td +++ b/test/testdrive/arrangement-sizes.td @@ -37,8 +37,8 @@ size bigint collection_timestamp "timestamp with time zone" hydration_complete boolean -# Set up a cluster with two replicas and arrangements large enough to -# clear the 10 MiB floor that `mz_object_arrangement_sizes` applies. +# Set up a cluster with two replicas and arrangements large enough that +# `mz_object_arrangement_sizes` reports a nonzero quantized size. > CREATE CLUSTER test SIZE '${arg.default-replica-size}', REPLICATION FACTOR 2 > SET cluster = test @@ -72,6 +72,21 @@ idx_t2 r2 true WHERE o.name = 'mv_filter' 0 +# Arrangements below the 5 MiB quantization threshold stay visible with a +# size of 0 rather than being dropped. +> CREATE TABLE small_t (a int, b text) +> INSERT INTO small_t SELECT g, repeat('x', 10) FROM generate_series(1, 100) g +> CREATE INDEX small_idx ON small_t (a) + +> SELECT o.name, r.name, s.size + FROM mz_internal.mz_object_arrangement_sizes s + JOIN mz_objects o ON o.id = s.object_id + JOIN mz_cluster_replicas r ON r.id = s.replica_id + WHERE o.name = 'small_idx' + ORDER BY r.name +small_idx r1 0 +small_idx r2 0 + # History table: at least one snapshot row per (object_id, replica_id) # pair, all with size > 0, and at least one row tagged # `hydration_complete = true` for each pair once arrangements settle. @@ -94,4 +109,19 @@ idx_t2 r2 true true true WHERE o.name IN ('idx_t', 'idx_t2') true +# The history skips size-0 rows: snapshots keep landing (fresh idx_t rows +# appear) while small_idx, live above with size 0, never gets a history row. +> SELECT count(*) > 0 + FROM mz_internal.mz_object_arrangement_size_history h + JOIN mz_objects o ON o.id = h.object_id + WHERE o.name = 'idx_t' + AND h.collection_timestamp > now() - INTERVAL '10 seconds' +true + +> SELECT count(*) + FROM mz_internal.mz_object_arrangement_size_history h + JOIN mz_objects o ON o.id = h.object_id + WHERE o.name = 'small_idx' +0 + > DROP CLUSTER test CASCADE