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/introspection.rs b/src/adapter/src/coord/introspection.rs index dc6743613332e..9a078c5c15dc4 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 { @@ -576,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 → @@ -610,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 017096579b00e..b3bf08dcff5f0 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,27 +461,47 @@ 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`. + /// + /// 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. /// - /// 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`. + /// 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(&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(); + // 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, @@ -490,67 +514,68 @@ 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, + &fresh_size_replicas, + &fresh_hydration_replicas, + ); + 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 +587,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 +1147,251 @@ 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`. +/// +/// 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)>, + fresh_size_replicas: &BTreeSet, + fresh_hydration_replicas: &BTreeSet, +) -> 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; + } + let replica_id = datums[HYDRATION_COL_REPLICA_ID].unwrap_str(); + if !fresh_hydration_replicas.contains(replica_id) { + continue; + } + hydrated.insert(( + replica_id.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 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 { + 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(); + 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 + // be misleading, so drop. + if size_datum.is_null() { + 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 { + 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"); + } + 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 \ + introspection data" + ); + } + records +} + +#[cfg(test)] +mod arrangement_sizes_records_tests { + use std::collections::BTreeSet; + + 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 + }, + ]) + } + + fn replicas(ids: &[&str]) -> BTreeSet { + ids.iter().map(|id| id.to_string()).collect() + } + + #[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 fresh = replicas(&["u1"]); + let records = arrangement_sizes_records(live, hydration, &fresh, &fresh); + 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 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_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. + 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/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/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": 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