diff --git a/doc/developer/design/20260827_durable_replica_hydration_history.md b/doc/developer/design/20260827_durable_replica_hydration_history.md new file mode 100644 index 0000000000000..b0f2f7fd3816a --- /dev/null +++ b/doc/developer/design/20260827_durable_replica_hydration_history.md @@ -0,0 +1,82 @@ +# Durable Replica Hydration History + +## Context + +The [durable object hydration history](20260817_durable_object_hydration_history.md) +records successful hydration for individual dataflows. This design extends that +collector with replica-wide hydration episodes and the resource high-water marks +visible when each episode is recorded. + +The extension reuses the object collector's scheduling, replica-targeted +read-then-write path, exact-timestamp OCC, retention, and migration protections. +This document describes only the replica-level additions. + +## Episode boundaries + +Each live non-transient compute export contributes an interval from its earliest +worker installation to its latest worker hydration, and counts as hydrated once +every visible worker has reported its finish. Transient query dataflows are +excluded because the collection query itself creates one. + +A replica episode is a connected component in the union of those export +intervals. Two intervals belong to one episode if they overlap directly or +through a chain of overlapping intervals. A gap means the replica was fully +hydrated before the next export was installed, so the next interval starts a +new episode. + +Each sweep records only the latest completed episode visible in its snapshot. +An export that has not hydrated keeps its episode in progress. Whether it is +slow or permanently stuck is unobservable, so an in-progress episode is not a +failure state: it is recorded when it completes and does not block earlier, +disconnected completed episodes. Episodes that complete between sweeps and +exports that retract before a sweep leave no evidence. The current inputs can +therefore record successful episodes only. Failed, canceled, and OOM-killed +outcomes need an additional durable replica signal. + +Process-local clocks stamp both interval endpoints. Clock skew can merge +episodes that did not overlap in real time. Once an episode is recorded, a +monotonic history guard prevents a later snapshot from interpreting retracted +intervals as an earlier or overlapping episode. + +## Resource interpretation + +`peak_memory_bytes` is the maximum `cgroup memory_peak` across replica +processes. `peak_disk_bytes` is the maximum sampled `statvfs fs_used_peak` when a +scratch filesystem is present. Otherwise it is the maximum kernel-maintained +`cgroup swap_peak`. + +Replica memory and disk limits apply independently to each process. The maximum +process peak therefore answers whether any process approached its limit. Adding +process maxima would combine peaks that may not have occurred simultaneously. + +The operating system's peaks cover the process lifetime through the collector's +observation. They are not bounded by `finished_at`, so work after hydration and +before collection can raise them. Later episodes can also include an earlier +high-water mark. A true episode peak requires a reset or a separately retained +interval maximum at the replica. + +The collector requires at least one resource observation from every configured +process before writing. Individual peak metrics can still be absent, which is +represented by `NULL` rather than a zero sentinel. + +## History table + +```text +mz_internal.mz_replica_hydration_history + replica_id text not null + cluster_id text not null + started_at timestamptz not null + finished_at timestamptz null + object_count uint8 not null + peak_memory_bytes uint8 null + peak_disk_bytes uint8 null + status text not null +``` + +An episode is identified operationally by `(replica_id, started_at)`. The table +does not declare a key or index. Collection runs on the selected replica, so a +catalog-server index would not avoid importing and arranging the history there. + +Rows currently have a populated `finished_at` and the status `hydrated`. +Resource columns are nullable because the available kernel and filesystem +observations depend on the replica platform. diff --git a/doc/user/content/reference/system-catalog/mz_internal.md b/doc/user/content/reference/system-catalog/mz_internal.md index 39fbf312341a4..e5054adc22cc0 100644 --- a/doc/user/content/reference/system-catalog/mz_internal.md +++ b/doc/user/content/reference/system-catalog/mz_internal.md @@ -743,6 +743,31 @@ logical timestamp, so the recorded finish can precede the latest process's finis | `hydrated_at` | [`timestamp with time zone`] | When hydration finished. | | `status` | [`text`] | The terminal status. Currently always `hydrated`. | +## `mz_replica_hydration_history` + +The `mz_replica_hydration_history` table records successful replica hydration +episodes. An episode begins when a maintained compute dataflow is installed on +a fully hydrated replica and finishes when every running maintained compute +dataflow has hydrated. + +By default, rows are retained for 30 days while collection is enabled. Recording +is best effort, and only the latest completed episode visible in each collection +is recorded. Resource peaks cover the replica processes' lifetimes through +collection, not only the hydration episode. On a multi-process replica, the +table records the largest peak reported by any process. + + +| Field | Type | Meaning | +| ------------------- | ---------------------------- | ------------------------------------------------------------------------------------------------------------------------ | +| `replica_id` | [`text`] | The ID of the cluster replica. May name a replica that no longer exists. | +| `cluster_id` | [`text`] | The ID of the replica's cluster. | +| `started_at` | [`timestamp with time zone`] | The earliest maintained compute dataflow installation in the hydration episode. | +| `finished_at` | [`timestamp with time zone`] | The latest maintained compute dataflow hydration in the hydration episode. | +| `object_count` | [`uint8`] | The number of maintained compute dataflows in the hydration episode. | +| `peak_memory_bytes` | [`uint8`] | The largest process-lifetime cgroup memory high-water mark reported by any process when the collector recorded the episode. `NULL` if the platform reports no cgroup memory peak. | +| `peak_disk_bytes` | [`uint8`] | The largest process-lifetime scratch-filesystem or swap high-water mark reported by any process when the collector recorded the episode. Filesystem peaks are sampled lower bounds. `NULL` if neither measurement is available. | +| `status` | [`text`] | The hydration episode's status. Currently always `hydrated`. | + ## `mz_object_transitive_dependencies` The `mz_object_transitive_dependencies` view describes the transitive dependency structure between diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index f8af94bdea893..57a326de24bea 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -963,7 +963,7 @@ metrics: source: src/adapter/src/metrics.rs visibility: internal - name: mz_hydration_history_retention_batch_full_total - help: Total hydration-history sweeps whose retention batch was full. Repeated increments mean retention may not be keeping up with its schedule. + help: Total hydration-history retention batches that were full. Repeated increments mean retention may not be keeping up with its schedule. source: src/adapter/src/metrics.rs visibility: internal - name: mz_hydration_history_rows_affected_total diff --git a/src/adapter-types/src/dyncfgs.rs b/src/adapter-types/src/dyncfgs.rs index 1f50dafbd0aad..c597767b9942e 100644 --- a/src/adapter-types/src/dyncfgs.rs +++ b/src/adapter-types/src/dyncfgs.rs @@ -429,11 +429,11 @@ pub const HYDRATION_HISTORY_COLLECTION_INTERVAL: Config = Config::new( ParameterScope::Environment, ); -/// How long to retain completed object hydration episodes. +/// How long to retain completed object and replica hydration episodes. pub const HYDRATION_HISTORY_RETENTION_PERIOD: Config = Config::new( "hydration_history_retention_period", Duration::from_hours(30 * 24), - "How long to retain rows in mz_internal.mz_object_hydration_history.", + "How long to retain rows in mz_internal.mz_object_hydration_history and mz_internal.mz_replica_hydration_history.", ParameterScope::Environment, ); diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index a96caf5281975..66394096d220d 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -39,6 +39,7 @@ use mz_catalog::builtin::{ BUILTIN_LOOKUP, Builtin, Fingerprint, MZ_CATALOG_RAW, MZ_CATALOG_RAW_DESCRIPTION, MZ_CLUSTER_REPLICA_FRONTIERS_DESCRIPTION, MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY_DESCRIPTION, MZ_OBJECT_HYDRATION_HISTORY, MZ_OBJECT_HYDRATION_HISTORY_DESCRIPTION, + MZ_REPLICA_HYDRATION_HISTORY, MZ_REPLICA_HYDRATION_HISTORY_DESCRIPTION, MZ_STORAGE_USAGE_BY_SHARD, MZ_STORAGE_USAGE_BY_SHARD_DESCRIPTION, RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL, }; @@ -689,18 +690,20 @@ fn participates_in_forced_migration( match builtin { // A forced replacement allocates a fresh shard, which discards the // table's contents. Exclude the tables whose contents are the point: - // storage usage is retained for billing, and hydration history cannot + // storage usage is retained for billing, and hydration histories cannot // be rebuilt from any other source. // - // Hydration history takes part in a forced `Evolution`, which keeps the - // rows. It has to: dev upgrades force one for every object, and a table - // left out of the plan never gets its new schema registered, so - // `update_fingerprints` panics at open as soon as the desc changes. See - // the tripwire in `validate_migration_steps` for how to give up the + // Hydration history tables take part in a forced `Evolution`, which + // keeps the rows. They have to: dev upgrades force one for every object, + // and a table left out of the plan never gets its new schema registered. + // `update_fingerprints` then panics at open as soon as the desc changes. + // See the tripwire in `validate_migration_steps` for how to give up the // replacement exemption deliberately. Table(table) => { **table != *MZ_STORAGE_USAGE_BY_SHARD - && (mechanism != Mechanism::Replacement || **table != *MZ_OBJECT_HYDRATION_HISTORY) + && (mechanism != Mechanism::Replacement + || (**table != *MZ_OBJECT_HYDRATION_HISTORY + && **table != *MZ_REPLICA_HYDRATION_HISTORY)) } MaterializedView(..) => true, Source(source) => **source != *MZ_CATALOG_RAW, @@ -819,6 +822,10 @@ impl Migration { &*MZ_OBJECT_HYDRATION_HISTORY_DESCRIPTION, object, "replacing mz_object_hydration_history clears it, see the comment above" ); + assert_ne!( + &*MZ_REPLICA_HYDRATION_HISTORY_DESCRIPTION, object, + "replacing mz_replica_hydration_history clears it, see the comment above" + ); } // `mz_catalog_raw` cannot be migrated because it contains the durable catalog and it diff --git a/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs b/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs index 2d168684d5f3f..d4ead4dcf08be 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration_tests.rs @@ -29,16 +29,21 @@ use super::*; #[mz_ore::test] fn hydration_history_forced_migration_policy() { - let hydration_history = Builtin::Table(&*MZ_OBJECT_HYDRATION_HISTORY); - - assert!(participates_in_forced_migration( - &hydration_history, - Mechanism::Evolution - )); - assert!(!participates_in_forced_migration( - &hydration_history, - Mechanism::Replacement - )); + for table in [ + &*MZ_OBJECT_HYDRATION_HISTORY, + &*MZ_REPLICA_HYDRATION_HISTORY, + ] { + let hydration_history = Builtin::Table(table); + + assert!(participates_in_forced_migration( + &hydration_history, + Mechanism::Evolution + )); + assert!(!participates_in_forced_migration( + &hydration_history, + Mechanism::Replacement + )); + } } #[test] // allow(test-attribute) diff --git a/src/adapter/src/coord.rs b/src/adapter/src/coord.rs index 721ba8975bcd4..f901312bd7df8 100644 --- a/src/adapter/src/coord.rs +++ b/src/adapter/src/coord.rs @@ -100,7 +100,7 @@ use mz_auth::password::Password; use mz_build_info::BuildInfo; use mz_catalog::builtin::{ BUILTINS, BUILTINS_STATIC, MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY, MZ_OBJECT_HYDRATION_HISTORY, - MZ_STORAGE_USAGE_BY_SHARD, + MZ_REPLICA_HYDRATION_HISTORY, MZ_STORAGE_USAGE_BY_SHARD, }; use mz_catalog::config::{AwsPrincipalContext, BuiltinItemMigrationConfig, ClusterReplicaSizeMap}; use mz_catalog::durable::OpenableDurableCatalogState; @@ -3186,6 +3186,8 @@ impl Coordinator { .resolve_builtin_table(&MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY), self.catalog() .resolve_builtin_table(&MZ_OBJECT_HYDRATION_HISTORY), + self.catalog() + .resolve_builtin_table(&MZ_REPLICA_HYDRATION_HISTORY), ]); let mut retraction_tasks = Vec::new(); @@ -4337,6 +4339,14 @@ impl Coordinator { } } } + + // The sweep can own timestamp-oracle senders through its background + // client. Release them before the coordinator runtime starts shutting + // down the oracle workers. + if let Some(sweep) = self.hydration_history_sweep.take() { + sweep.abort_and_wait().await; + } + // Try and cleanup as a best effort. There may be some async tasks out there holding a // reference that prevents us from cleaning up. if let Some(catalog) = Arc::into_inner(self.catalog) { diff --git a/src/adapter/src/coord/hydration_history.rs b/src/adapter/src/coord/hydration_history.rs index b7362dff9be09..6ef646717f5b9 100644 --- a/src/adapter/src/coord/hydration_history.rs +++ b/src/adapter/src/coord/hydration_history.rs @@ -7,12 +7,12 @@ // the Business Source License, use of this software will be governed // by the Apache License, Version 2.0. -//! Durable history collection for completed compute-object hydration episodes. +//! Durable history collection for completed object and replica hydration episodes. //! //! One sweep visits a single user replica, installs a replica-targeted //! subscribe that diffs that replica's live hydration timestamps against the -//! durable history table, and appends what is missing through the timestamped -//! OCC write path. Including the history table in the read expression is what +//! durable history tables, and appends what is missing through the timestamped +//! OCC write path. Including each history table in its read expression is what //! makes the write idempotent across concurrent `environmentd` processes: two //! collectors that compute the same row race for one write timestamp, and the //! loser observes the winner's append through its own subscribe and finds @@ -22,10 +22,10 @@ //! replicas revisits each one approximately every `N * interval`. Lowering the //! interval improves freshness at the cost of more replica dataflow installs. //! -//! Collection is sampling, not an event log. An episode whose live row is -//! retracted before its replica's turn in the sweep (a dropped object, or a -//! replica that restarts first) is not recorded, and cannot be, because -//! the only evidence is gone. See the design doc for why that is accepted here. +//! Collection is sampling, not an event log. Replica history records only the +//! latest completed episode visible in a sweep. Intermediate episodes and +//! intervals retracted before collection leave no evidence and are not recorded. +//! See the design doc for the resulting semantics. use std::collections::BTreeMap; use std::sync::Arc; @@ -36,7 +36,9 @@ use mz_adapter_types::dyncfgs::{ FRONTEND_READ_THEN_WRITE, HYDRATION_HISTORY_COLLECTION_INTERVAL, HYDRATION_HISTORY_RETENTION_PERIOD, }; -use mz_catalog::builtin::{MZ_CATALOG_SERVER_CLUSTER, MZ_OBJECT_HYDRATION_HISTORY}; +use mz_catalog::builtin::{ + MZ_CATALOG_SERVER_CLUSTER, MZ_OBJECT_HYDRATION_HISTORY, MZ_REPLICA_HYDRATION_HISTORY, +}; use mz_cluster_client::ReplicaId; use mz_controller::clusters::{ClusterStatus, ReplicaLocation}; use mz_controller_types::ClusterId; @@ -208,8 +210,12 @@ impl Coordinator { // used by tests. Their bounded mutation determines readiness. ReplicaLocation::Unmanaged(_) => true, }) - .map(|replica| (replica.cluster_id, replica.replica_id)) - .sorted_by_key(|(_, replica_id)| *replica_id) + .map(|replica| ReplicaTarget { + cluster_id: replica.cluster_id, + replica_id: replica.replica_id, + process_count: replica.config.location.num_processes(), + }) + .sorted_by_key(|replica| replica.replica_id) .collect_vec(); let catalog = self.owned_catalog(); @@ -223,16 +229,16 @@ impl Coordinator { .map(|replica| (catalog_server.id, replica.replica_id)); let replica = next_replica(&replicas, self.hydration_history_replica_cursor); - if let Some((_, replica_id)) = replica { - self.hydration_history_replica_cursor = Some(replica_id); + if let Some(replica) = replica { + self.hydration_history_replica_cursor = Some(replica.replica_id); } let mut sweep = self.new_sweep(catalog, retention); let internal_cmd_tx = self.internal_cmd_tx.clone(); let handle = task::spawn(|| "hydration_history_sweep", async move { let started = Instant::now(); - if let Some((cluster_id, replica_id)) = replica { - sweep.collect(cluster_id, replica_id).await; + if let Some(replica) = replica { + sweep.collect(replica).await; } // Retention runs even when collection failed above. A replica that @@ -279,7 +285,8 @@ impl Coordinator { ); Sweep { client, - history_id: catalog.resolve_builtin_table(&MZ_OBJECT_HYDRATION_HISTORY), + object_history_id: catalog.resolve_builtin_table(&MZ_OBJECT_HYDRATION_HISTORY), + replica_history_id: catalog.resolve_builtin_table(&MZ_REPLICA_HYDRATION_HISTORY), catalog, metrics: self.metrics.clone(), wall_time: self.now_datetime(), @@ -288,18 +295,23 @@ impl Coordinator { } } +/// A user replica eligible for one collection step. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct ReplicaTarget { + cluster_id: ClusterId, + replica_id: ReplicaId, + process_count: usize, +} + /// Picks the replica after `cursor`, wrapping around at the end. /// /// `replicas` must be sorted ascending by replica id. Unsorted input still /// returns a replica but degenerates the rotation, revisiting some replicas and /// starving others. -fn next_replica( - replicas: &[(ClusterId, ReplicaId)], - cursor: Option, -) -> Option<(ClusterId, ReplicaId)> { +fn next_replica(replicas: &[ReplicaTarget], cursor: Option) -> Option { replicas .iter() - .find(|(_, replica_id)| cursor.is_none_or(|cursor| *replica_id > cursor)) + .find(|replica| cursor.is_none_or(|cursor| replica.replica_id > cursor)) .or_else(|| replicas.first()) .copied() } @@ -345,7 +357,7 @@ fn next_replica( /// not-yet-recorded dataflow, and the OCC path rejects a result that exceeds /// `max_result_size` or `max_query_result_size`. At their 1 GiB defaults that /// ceiling only matters at millions of dataflows per replica. -fn collect_sql(cluster_id: ClusterId, replica_id: ReplicaId, cutoff: &str) -> String { +fn object_collection_sql(cluster_id: ClusterId, replica_id: ReplicaId, cutoff: &str) -> String { // Interpolating into SQL is safe here: the ids are catalog-internal and the // cutoff is an RFC 3339 timestamp we formatted ourselves. Nothing in this // query comes from a user. @@ -394,12 +406,168 @@ fn collect_sql(cluster_id: ClusterId, replica_id: ReplicaId, cutoff: &str) -> St ) } +/// Returns SQL for the latest completed compute hydration episode and its +/// process resource peaks. +/// +/// Episodes are connected components of export hydration intervals. An export +/// that has not hydrated keeps its component open: whether it is slow or +/// permanently stuck is unobservable, so an open component is simply an +/// in-progress episode, recorded when (if) it completes. It blocks only its +/// own component: an unhydrated export installed at or before a completed +/// component's finish would extend that component, one installed later belongs +/// to a later episode. The latest completed component disconnected from every +/// open one is recorded. The monotonic history guard still admits an open +/// episode once it completes, because its start lies after every recorded +/// finish. Cross-process clock skew can break that ordering, in which case the +/// guard suppresses the episode rather than misrecording it. +/// +/// Collection also waits until every configured replica process has reported +/// resource usage. The query itself narrates how each step works. +fn replica_collection_sql(target: ReplicaTarget, cutoff: &str) -> String { + let ReplicaTarget { + cluster_id, + replica_id, + process_count, + } = target; + // Interpolating into SQL is safe here: the ids and process count are + // catalog-internal and the cutoff is an RFC 3339 timestamp we formatted. + format!( + "WITH + -- One hydration interval per compute export: earliest install and + -- latest finish across its workers. Hydrated only once every worker + -- visible at this timestamp has finished. + objects AS ( + SELECT + t.export_id AS object_id, + min(t.installed_at) AS installed_at, + max(t.hydrated_at) AS hydrated_at, + count(*) = count(t.hydrated_at) AS hydrated + FROM mz_introspection.mz_compute_hydration_times_per_worker AS t + WHERE t.export_id NOT LIKE 't%' + GROUP BY t.export_id + ), + -- Completed intervals in install order, each with the coverage + -- horizon: the latest finish among this and all earlier intervals. + covered AS ( + SELECT + object_id, + installed_at, + hydrated_at, + max(hydrated_at) OVER ( + ORDER BY installed_at, object_id + ROWS UNBOUNDED PRECEDING + ) AS covered_through + FROM objects + WHERE hydrated + ), + -- An interval starts a new episode when the horizon just before it + -- does not reach its install: for a moment, nothing was hydrating. + flagged AS ( + SELECT + object_id, + installed_at, + hydrated_at, + lag(covered_through) OVER ( + ORDER BY installed_at, object_id + ) IS NULL + OR lag(covered_through) OVER ( + ORDER BY installed_at, object_id + ) < installed_at AS starts_episode + FROM covered + ), + -- Each interval belongs to the latest episode start at or before it. + labeled AS ( + SELECT + installed_at, + hydrated_at, + max(CASE WHEN starts_episode THEN installed_at END) OVER ( + ORDER BY installed_at, object_id + ROWS UNBOUNDED PRECEDING + ) AS episode_started_at + FROM flagged + ), + -- One row per completed episode. + episodes AS ( + SELECT + episode_started_at AS started_at, + max(hydrated_at) AS finished_at, + count(*)::uint8 AS object_count + FROM labeled + GROUP BY episode_started_at + ), + -- The earliest install of an export that has not hydrated yet. + open_min AS ( + SELECT min(installed_at) AS v FROM objects WHERE NOT hydrated + ), + -- The episode to record: the latest one that finished before any + -- unhydrated export was installed. An episode finishing at or after + -- open_min contains that open interval and is still in progress. + -- Comparing against this one scalar, instead of joining episodes + -- with open intervals, avoids a cross product that is quadratic when + -- many episodes coexist with many still-hydrating exports. + episode AS ( + SELECT e.started_at, e.finished_at, e.object_count + FROM episodes AS e, open_min AS o + WHERE o.v IS NULL OR e.finished_at < o.v + ORDER BY e.started_at DESC + LIMIT 1 + ), + -- Process-lifetime resource high-water marks, and how many processes + -- have reported them. + resources AS ( + SELECT + count(DISTINCT process_id) AS process_count, + max(value) FILTER ( + WHERE source = 'cgroup' AND metric = 'memory_peak' + ) AS peak_memory_bytes, + coalesce( + max(value) FILTER ( + WHERE source = 'statvfs' AND metric = 'fs_used_peak' + ), + max(value) FILTER ( + WHERE source = 'cgroup' AND metric = 'swap_peak' + ) + ) AS peak_disk_bytes + FROM mz_introspection.mz_cluster_replica_resource_usage + ), + -- The history row to write, held back until every configured process + -- has reported resource usage and dropped once the episode has aged + -- past the retention cutoff. + candidate AS ( + SELECT + '{replica_id}'::text AS replica_id, + '{cluster_id}'::text AS cluster_id, + e.started_at, + e.finished_at, + e.object_count, + r.peak_memory_bytes, + r.peak_disk_bytes, + 'hydrated'::text AS status + FROM episode AS e + CROSS JOIN resources AS r + WHERE r.process_count = {process_count}::uint8 + AND e.finished_at >= TIMESTAMPTZ '{cutoff}' + ) + -- Skip episodes the history already covers: a recorded row finishing + -- at or after this start is this episode, or overlaps it under + -- cross-process clock skew. + SELECT c.* + FROM candidate AS c + WHERE NOT EXISTS ( + SELECT 1 + FROM mz_internal.mz_replica_hydration_history AS h + WHERE h.replica_id = c.replica_id + AND h.finished_at >= c.started_at + )" + ) +} + /// A bounded batch of history rows that have aged out. /// /// Only rows with a `hydrated_at` age out. Every row written today has one, and /// a row without one would be immortal here, so an unfinished-episode /// representation needs a second age basis before it can be recorded. -fn retention_sql(cutoff: &str) -> String { +fn object_retention_sql(cutoff: &str) -> String { // The LIMIT has to sit inside a subquery. A top-level LIMIT lands in the // plan's `RowSetFinishing`, which this OCC stage cannot apply. Inside a // derived table it lowers into the relation expression instead. @@ -416,25 +584,60 @@ fn retention_sql(cutoff: &str) -> String { ) } +/// A bounded batch of replica history rows that have aged out. +fn replica_retention_sql(cutoff: &str) -> String { + format!( + "SELECT * FROM ( + SELECT + replica_id, cluster_id, started_at, finished_at, object_count, + peak_memory_bytes, peak_disk_bytes, status + FROM mz_internal.mz_replica_hydration_history + WHERE finished_at < TIMESTAMPTZ '{cutoff}' + ORDER BY finished_at + LIMIT {RETENTION_BATCH_SIZE} + )" + ) +} + /// What one sweep needs to run its mutations against the history table. struct Sweep { client: PeekClient, catalog: Arc, - history_id: CatalogItemId, + object_history_id: CatalogItemId, + replica_history_id: CatalogItemId, metrics: Metrics, wall_time: chrono::DateTime, - /// Rows finishing before this have aged out. Both steps apply it, so a live - /// log row cannot resurrect an episode retention just retracted. + /// Rows finishing before this have aged out. Both steps apply it, so this + /// sweep cannot resurrect an episode its own retention step retracts. + /// Concurrent sweeps can have different cutoffs, making retention eventual. cutoff: String, } impl Sweep { - /// Appends this replica's completed episodes that the table is missing. - async fn collect(&mut self, cluster_id: ClusterId, replica_id: ReplicaId) { - let sql = collect_sql(cluster_id, replica_id, &self.cutoff); + /// Appends completed object and replica episodes from one replica. + async fn collect(&mut self, target: ReplicaTarget) { + let ReplicaTarget { + cluster_id, + replica_id, + .. + } = target; + let sql = object_collection_sql(cluster_id, replica_id, &self.cutoff); let _ = self .run( "collection", + self.object_history_id, + cluster_id, + replica_id, + MutationKind::Insert, + &sql, + ) + .await; + + let sql = replica_collection_sql(target, &self.cutoff); + let _ = self + .run( + "replica_collection", + self.replica_history_id, cluster_id, replica_id, MutationKind::Insert, @@ -445,20 +648,35 @@ impl Sweep { /// Retracts one bounded batch of aged-out rows. async fn retain(&mut self, cluster_id: ClusterId, replica_id: ReplicaId) { - let sql = retention_sql(&self.cutoff); - let Some(deleted) = self + let sql = object_retention_sql(&self.cutoff); + if let Some(deleted) = self .run( "retention", + self.object_history_id, cluster_id, replica_id, MutationKind::Delete, &sql, ) .await - else { - return; - }; - if deleted == RETENTION_BATCH_SIZE { + && deleted == RETENTION_BATCH_SIZE + { + self.metrics.hydration_history_retention_batch_full.inc(); + } + + let sql = replica_retention_sql(&self.cutoff); + if let Some(deleted) = self + .run( + "replica_retention", + self.replica_history_id, + cluster_id, + replica_id, + MutationKind::Delete, + &sql, + ) + .await + && deleted == RETENTION_BATCH_SIZE + { self.metrics.hydration_history_retention_batch_full.inc(); } } @@ -476,13 +694,14 @@ impl Sweep { async fn run( &mut self, step: &'static str, + history_id: CatalogItemId, cluster_id: ClusterId, replica_id: ReplicaId, kind: MutationKind, sql: &str, ) -> Option { let mutation = async { - let plan = plan_mutation(&self.catalog, self.history_id, kind, sql)?; + let plan = plan_mutation(&self.catalog, history_id, kind, sql)?; let mut session = Session::dummy(); session.start_transaction_single_stmt(self.wall_time); let response = self @@ -521,7 +740,9 @@ impl Sweep { } Ok(Err(error)) => { self.observe_mutation(step, "error"); - if step == "collection" && matches!(&error, AdapterError::ReadThenWriteContention) { + if step.ends_with("collection") + && matches!(&error, AdapterError::ReadThenWriteContention) + { warn!( %step, %cluster_id, %replica_id, %error, "hydration history step failed, the replica's introspection frontier \ @@ -535,7 +756,7 @@ impl Sweep { // A trailing replica can repeatedly certify a target only after the // oracle has advanced past it. Each refused write raises the target, // and the conflict loop can continue until this timeout fires. - Err(_) if step == "collection" => { + Err(_) if step.ends_with("collection") => { self.observe_mutation(step, "timeout"); warn!( %step, %cluster_id, %replica_id, @@ -630,7 +851,18 @@ mod tests { #[mz_ore::test] fn replica_sweep_advances_and_wraps() { let cluster = ClusterId::user(1).expect("valid cluster ID"); - let replicas = [(cluster, ReplicaId::User(1)), (cluster, ReplicaId::User(3))]; + let replicas = [ + ReplicaTarget { + cluster_id: cluster, + replica_id: ReplicaId::User(1), + process_count: 1, + }, + ReplicaTarget { + cluster_id: cluster, + replica_id: ReplicaId::User(3), + process_count: 1, + }, + ]; assert_eq!(next_replica(&replicas, None), Some(replicas[0])); assert_eq!( @@ -696,7 +928,7 @@ mod tests { #[mz_ore::test] fn collect_requires_every_worker() { let cutoff = "1970-01-01T00:00:00+00:00"; - let sql = collect_sql( + let sql = object_collection_sql( ClusterId::user(1).expect("valid cluster ID"), ReplicaId::User(2), cutoff, @@ -717,4 +949,78 @@ mod tests { "{sql}" ); } + + /// Replica episodes are connected components of object hydration intervals, + /// enumerated gaps-and-islands style. An episode still connected to an open + /// interval is skipped via a scalar comparison against the earliest open + /// install, and the latest remaining episode is recorded. The query must + /// also wait for every replica process before it snapshots process-local + /// high-water marks. + #[mz_ore::test] + fn replica_collection_uses_latest_completed_interval_island() { + let sql = replica_collection_sql( + ReplicaTarget { + cluster_id: ClusterId::user(1).expect("valid cluster ID"), + replica_id: ReplicaId::User(2), + process_count: 3, + }, + "1970-01-01T00:00:00+00:00", + ); + let normalized_sql = sql.split_whitespace().collect::>().join(" "); + + // The gaps-and-islands scaffolding: running coverage horizon, gap + // detection against the previous row's horizon, episode labels, and + // per-episode aggregation. + assert!(sql.contains("ROWS UNBOUNDED PRECEDING"), "{sql}"); + assert!(sql.contains("lag(covered_through)"), "{sql}"); + assert!( + sql.contains("CASE WHEN starts_episode THEN installed_at END"), + "{sql}" + ); + assert!(sql.contains("GROUP BY episode_started_at"), "{sql}"); + // The unfinished-export guard applies per episode. A replica-wide + // all-hydrated gate would lose a completed episode for good: once the + // in-progress one finishes, it is the latest and the earlier one is + // never recorded. + assert!(!sql.contains("bool_and(hydrated)"), "{sql}"); + // The guard compares each episode against the earliest open install, + // one scalar row. A join against all open intervals is quadratic when + // many episodes coexist with many still-hydrating exports. + assert!( + normalized_sql.contains("WHERE o.v IS NULL OR e.finished_at < o.v"), + "{sql}" + ); + assert!(!sql.contains("o.installed_at <= e.finished_at"), "{sql}"); + assert!( + normalized_sql.contains("ORDER BY e.started_at DESC LIMIT 1"), + "{sql}" + ); + assert!(sql.contains("r.process_count = 3::uint8"), "{sql}"); + assert!(sql.contains("WHERE t.export_id NOT LIKE 't%'"), "{sql}"); + assert!(!sql.contains("WHERE t.export_id LIKE 'u%'"), "{sql}"); + assert!(!sql.contains("mz_object_global_ids"), "{sql}"); + assert!(!sql.contains("mz_catalog.mz_objects"), "{sql}"); + assert!( + normalized_sql + .contains("max(value) FILTER ( WHERE source = 'cgroup' AND metric = 'memory_peak'"), + "{sql}" + ); + assert!( + normalized_sql.contains( + "max(value) FILTER ( WHERE source = 'statvfs' AND metric = 'fs_used_peak'" + ), + "{sql}" + ); + assert!( + normalized_sql + .contains("max(value) FILTER ( WHERE source = 'cgroup' AND metric = 'swap_peak'"), + "{sql}" + ); + assert!(!normalized_sql.contains("sum(value)"), "{sql}"); + assert!( + sql.contains("FROM mz_internal.mz_replica_hydration_history"), + "{sql}" + ); + assert!(sql.contains("h.finished_at >= c.started_at"), "{sql}"); + } } diff --git a/src/adapter/src/metrics.rs b/src/adapter/src/metrics.rs index 5cc8492f32960..a2becbeacec5a 100644 --- a/src/adapter/src/metrics.rs +++ b/src/adapter/src/metrics.rs @@ -150,7 +150,7 @@ impl Metrics { )), hydration_history_retention_batch_full: registry.register(metric!( name: "mz_hydration_history_retention_batch_full_total", - help: "Total hydration-history sweeps whose retention batch was full. Repeated increments mean retention may not be keeping up with its schedule.", + help: "Total hydration-history retention batches that were full. Repeated increments mean retention may not be keeping up with its schedule.", )), hydration_history_rows_affected: registry.register(metric!( name: "mz_hydration_history_rows_affected_total", diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs index a3dbda548f476..fe7c69e87020f 100644 --- a/src/catalog/src/builtin.rs +++ b/src/catalog/src/builtin.rs @@ -845,6 +845,14 @@ pub static MZ_OBJECT_HYDRATION_HISTORY_DESCRIPTION: LazyLock = + LazyLock::new(|| SystemObjectDescription { + schema_name: MZ_REPLICA_HYDRATION_HISTORY.schema.to_string(), + object_type: CatalogItemType::Table, + object_name: MZ_REPLICA_HYDRATION_HISTORY.name.to_string(), + }); + /// Identifies [`MZ_CLUSTER_REPLICA_FRONTIERS`] for the schema-migration guard in /// `builtin_schema_migration.rs`, which forbids migrating this source because the 0dt /// caught-up gate reads the leader's shard for it to learn the live frontiers. @@ -1487,6 +1495,7 @@ pub static BUILTINS_STATIC: LazyLock>> = LazyLock::ne Builtin::View(&MZ_MCP_DATA_PRODUCTS), Builtin::View(&MZ_MCP_DATA_PRODUCT_DETAILS), Builtin::Table(&MZ_OBJECT_HYDRATION_HISTORY), + Builtin::Table(&MZ_REPLICA_HYDRATION_HISTORY), ]; builtin_items.extend(notice::builtins()); diff --git a/src/catalog/src/builtin/mz_internal.rs b/src/catalog/src/builtin/mz_internal.rs index 463df5d4168cd..90355098b0a67 100644 --- a/src/catalog/src/builtin/mz_internal.rs +++ b/src/catalog/src/builtin/mz_internal.rs @@ -5242,6 +5242,114 @@ pub static MZ_OBJECT_HYDRATION_HISTORY: LazyLock = LazyLock::new(| }), }); +/// Successful hydration episodes for cluster replicas. +/// +/// Exempt from the bootstrap reset and from forced shard replacement, since the +/// contents cannot be rebuilt from anything else. Schema evolution keeps them and +/// applies normally. Clearing them for a schema change is still allowed, see the +/// tripwire in `validate_migration_steps`. +pub static MZ_REPLICA_HYDRATION_HISTORY: LazyLock = LazyLock::new(|| BuiltinTable { + name: "mz_replica_hydration_history", + schema: MZ_INTERNAL_SCHEMA, + oid: oid::TABLE_MZ_REPLICA_HYDRATION_HISTORY_OID, + desc: RelationDesc::builder() + .with_column("replica_id", SqlScalarType::String.nullable(false)) + .with_column("cluster_id", SqlScalarType::String.nullable(false)) + .with_column( + "started_at", + SqlScalarType::TimestampTz { precision: None }.nullable(false), + ) + .with_column( + "finished_at", + SqlScalarType::TimestampTz { precision: None }.nullable(true), + ) + .with_column("object_count", SqlScalarType::UInt64.nullable(false)) + .with_column("peak_memory_bytes", SqlScalarType::UInt64.nullable(true)) + .with_column("peak_disk_bytes", SqlScalarType::UInt64.nullable(true)) + .with_column("status", SqlScalarType::String.nullable(false)) + .finish(), + column_comments: BTreeMap::from_iter([ + ( + "replica_id", + "The ID of the cluster replica. May name a replica that no longer exists.", + ), + ("cluster_id", "The ID of the replica's cluster."), + ( + "started_at", + "The earliest maintained compute dataflow installation in the hydration episode.", + ), + ( + "finished_at", + "The latest maintained compute dataflow hydration in the hydration episode.", + ), + ( + "object_count", + "The number of maintained compute dataflows in the hydration episode.", + ), + ( + "peak_memory_bytes", + "The largest process-lifetime cgroup memory high-water mark reported by any process when the collector recorded the episode. `NULL` if the platform reports no cgroup memory peak.", + ), + ( + "peak_disk_bytes", + "The largest process-lifetime scratch-filesystem or swap high-water mark reported by any process when the collector recorded the episode. Filesystem peaks are sampled lower bounds. `NULL` if neither measurement is available.", + ), + ( + "status", + "The hydration episode's status. Currently always `hydrated`.", + ), + ]), + // Not a retained-metrics object: that would pin a 30 day compaction window, + // and our history lives in the rows, which the retention sweep retracts on + // its own schedule. Nothing reads this table at an old timestamp. + is_retained_metrics_object: false, + access: vec![PUBLIC_SELECT], + ontology: Some(Ontology { + entity_name: "replica_hydration_episode", + description: "Successful hydration episode on a cluster replica", + links: &const { + [ + OntologyLink { + name: "hydrated_on_cluster", + target: "cluster", + properties: LinkProperties::ForeignKey { + source_column: "cluster_id", + target_column: "id", + cardinality: Cardinality::ManyToOne, + source_id_type: None, + requires_mapping: None, + nullable: false, + note: Some( + "Hydration samples can outlive their cluster, so this reference may not resolve.", + ), + extra_key_columns: None, + }, + }, + OntologyLink { + name: "hydrated_on_replica", + target: "replica", + properties: LinkProperties::ForeignKey { + source_column: "replica_id", + target_column: "id", + cardinality: Cardinality::ManyToOne, + source_id_type: Some(mz_repr::SemanticType::ReplicaId), + requires_mapping: None, + nullable: false, + note: Some( + "Hydration samples can outlive their replica, so this reference may not resolve.", + ), + extra_key_columns: None, + }, + }, + ] + }, + column_semantic_types: &[ + ("replica_id", SemanticType::ReplicaId), + ("cluster_id", SemanticType::ClusterId), + ], + }), +}); + pub static MZ_COMPUTE_HYDRATION_STATUSES: LazyLock = LazyLock::new(|| BuiltinView { name: "mz_compute_hydration_statuses", schema: MZ_INTERNAL_SCHEMA, diff --git a/src/ore/src/task.rs b/src/ore/src/task.rs index 449e11fb037bb..0ce74ccdfdbc2 100644 --- a/src/ore/src/task.rs +++ b/src/ore/src/task.rs @@ -54,11 +54,17 @@ use tokio::task::{self, JoinHandle as TokioJoinHandle}; pub struct AbortOnDropHandle(JoinHandle); impl AbortOnDropHandle { - /// Checks if the task associated with this [`AbortOnDropHandle`] has finished.a + /// Checks if the task associated with this [`AbortOnDropHandle`] has finished. pub fn is_finished(&self) -> bool { self.0.inner.is_finished() } + /// Aborts the task, then waits for it to release its owned resources. + pub async fn abort_and_wait(mut self) { + self.0.inner.abort(); + let _ = (&mut self.0.inner).await; + } + // Note: adding an `abort(&self)` method here is incorrect; see the comment in JoinHandle::poll. } @@ -442,3 +448,38 @@ impl JoinSetExt for tokio::task::JoinSet { } } } + +#[cfg(test)] +mod tests { + use std::future; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + + use tokio::sync::oneshot; + + struct SetOnDrop(Arc); + + impl Drop for SetOnDrop { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + #[mz_ore::test(tokio::test)] + async fn abort_on_drop_handle_waits_for_task_drop() { + let dropped = Arc::new(AtomicBool::new(false)); + let guard = SetOnDrop(Arc::clone(&dropped)); + let (started_tx, started_rx) = oneshot::channel(); + let handle = super::spawn(|| "abort_on_drop_handle_waits_for_task_drop", async move { + let _guard = guard; + started_tx.send(()).expect("receiver remains live"); + future::pending::<()>().await; + }) + .abort_on_drop(); + + started_rx.await.expect("task starts"); + handle.abort_and_wait().await; + + assert!(dropped.load(Ordering::SeqCst)); + } +} diff --git a/src/pgrepr-consts/src/oid.rs b/src/pgrepr-consts/src/oid.rs index aa820195e6f1a..25eb0d8f7c98b 100644 --- a/src/pgrepr-consts/src/oid.rs +++ b/src/pgrepr-consts/src/oid.rs @@ -830,3 +830,4 @@ pub const MV_MZ_METRIC_SINKS_OID: u32 = 17120; pub const INDEX_MZ_METRIC_SINKS_IND_OID: u32 = 17121; pub const TABLE_MZ_OBJECT_HYDRATION_HISTORY_OID: u32 = 17122; pub const LOG_MZ_CLUSTER_REPLICA_RESOURCE_USAGE_OID: u32 = 17123; +pub const TABLE_MZ_REPLICA_HYDRATION_HISTORY_OID: u32 = 17124; diff --git a/test/restart/mzcompose.py b/test/restart/mzcompose.py index c8b2dd56d3a80..2d4ee60222f9e 100644 --- a/test/restart/mzcompose.py +++ b/test/restart/mzcompose.py @@ -16,6 +16,7 @@ import copy import json import time +from datetime import datetime from textwrap import dedent import requests @@ -1437,13 +1438,16 @@ def wait_for(sql: str, expected: list[tuple], what: str) -> None: def workflow_hydration_history_survives_restart(c: Composition) -> None: - """`mz_object_hydration_history` rows outlive the process that wrote them. + """Durable object and replica hydration rows outlive their writer. Killing the service also restarts clusterd, so the replica hydrates again - and legitimately records a *second* episode with a fresh `installed_at`. - What must hold is that the pre-restart episode is still there afterwards, - unchanged, and that repeated sweeps do not duplicate it. Asserting a total - row count of one would instead assert that rehydration goes unrecorded. + and legitimately records fresh episodes: one when rehydration forms a + single episode, more when the introspection indexes complete before the + user dataflows install. What must hold is that the pre-restart episodes + are still there afterwards, unchanged, that every fresh episode starts + after every pre-restart finish, and that repeated sweeps do not duplicate + anything. Asserting exact row counts would instead assert on collection + timing. """ def episodes(name: str = "hydration_history_i") -> list[list]: @@ -1456,6 +1460,23 @@ def episodes(name: str = "hydration_history_i") -> list[list]: WHERE o.name = '{name}' ORDER BY h.installed_at""") + def replica_episodes() -> list[list]: + return c.sql_query(""" + SELECT h.replica_id, h.started_at::text, h.finished_at::text, + h.object_count::text, h.peak_memory_bytes::text, + h.peak_disk_bytes::text, h.status + FROM mz_internal.mz_replica_hydration_history AS h + JOIN mz_catalog.mz_cluster_replicas AS r ON r.id = h.replica_id + JOIN mz_catalog.mz_clusters AS c ON c.id = h.cluster_id + WHERE r.name = 'r1' AND c.name = 'hydration_history' + ORDER BY h.started_at""") + + def replica_episode_identities(episodes: list[list]) -> list[tuple[str, str]]: + return [(episode[0], episode[1]) for episode in episodes] + + def parse_ts(text: str) -> datetime: + return datetime.fromisoformat(text) + c.down(destroy_volumes=True) with c.override( Materialized( @@ -1492,10 +1513,27 @@ def episodes(name: str = "hydration_history_i") -> list[list]: len(before) == 1 ), f"expected exactly one episode, got {before} (empty means it timed out)" - # Discover which MV's persist-sink worker is off worker 0 instead of - # predicting it from user-ID allocation and hashing. Enough input data - # separates compute completion from the active worker's durable write. deadline = time.time() + 120 + replica_before = [] + while time.time() < deadline: + replica_before = replica_episodes() + if replica_before: + break + time.sleep(0.5) + assert replica_before, "replica hydration history timed out before restart" + + # Discover an MV whose persist-sink worker is off worker 0 instead of + # predicting it from user-ID allocation and hashing. Enough input data + # separates compute completion from the active worker's durable write, + # but a single MV is not a reliable trial: its sink can land on worker + # 0, and a fast snapshot write can collapse both workers' stamps into + # one logging batch. Either way the separation is unobservable on that + # MV, so once a trial is fully hydrated and disqualified, create + # another MV: a fresh id rolls the sink worker and a fresh snapshot + # write rolls the timing. + deadline = time.time() + 240 + mv_names = ["hydration_history_mv_a", "hydration_history_mv_b"] + max_mvs = 8 candidates = [] worker_rows = [] with c.sql_cursor(reuse_connection=True) as cursor: @@ -1503,7 +1541,8 @@ def episodes(name: str = "hydration_history_i") -> list[list]: cursor.execute("SET cluster = hydration_history") cursor.execute("SET cluster_replica = r1") while time.time() < deadline: - cursor.execute(""" + name_list = ", ".join(f"'{name}'" for name in mv_names) + cursor.execute(f""" SELECT mv.name, max(h.hydrated_at)::text, @@ -1513,15 +1552,12 @@ def episodes(name: str = "hydration_history_i") -> list[list]: JOIN mz_internal.mz_object_global_ids AS g ON g.global_id = h.export_id JOIN mz_catalog.mz_materialized_views AS mv ON mv.id = g.id - WHERE mv.name IN ( - 'hydration_history_mv_a', - 'hydration_history_mv_b' - ) + WHERE mv.name IN ({name_list}) GROUP BY mv.name HAVING count(*) = 2 AND count(*) = count(h.hydrated_at) ORDER BY mv.name - """) + """.encode()) worker_rows = cursor.fetchall() candidates = [ row @@ -1530,13 +1566,22 @@ def episodes(name: str = "hydration_history_i") -> list[list]: ] if candidates: break + if len(worker_rows) == len(mv_names) and len(mv_names) < max_mvs: + name = f"hydration_history_mv_{chr(ord('a') + len(mv_names))}" + cursor.execute(f""" + CREATE MATERIALIZED VIEW {name} + IN CLUSTER hydration_history + AS SELECT a + {len(mv_names) + 1} AS a + FROM hydration_history_t + """.encode()) + mv_names.append(name) time.sleep(0.5) finally: cursor.execute("RESET cluster_replica") cursor.execute("RESET cluster") assert candidates, ( - "test fixture did not produce an MV with its persist-sink worker " - f"off worker 0: {worker_rows}" + "no MV produced an observable off-worker-0 persist-sink finish, " + f"tried {len(mv_names)}: {worker_rows}" ) mv_name, latest_worker_finish, worker_zero_finish = candidates[0] @@ -1558,6 +1603,32 @@ def episodes(name: str = "hydration_history_i") -> list[list]: f"worker 0 finished at {worker_zero_finish}" ) + # Re-read the pre-restart episodes immediately before the kill. The + # waits above leave minutes in which a further legitimate episode can + # be recorded, and the fresh-episode assertions after the restart + # assume this set is current. Two equal consecutive reads shrink the + # remaining window to a sweep that starts and commits within one poll + # gap. + deadline = time.time() + 120 + replica_before = replica_episodes() + replica_before_settled = False + while time.time() < deadline: + time.sleep(2.0) + current = replica_episodes() + replica_before_settled = current == replica_before + if replica_before_settled: + break + replica_before = current + assert ( + replica_before_settled + ), f"pre-restart replica episodes did not settle: {replica_before}" + replica_before_ids = replica_episode_identities(replica_before) + replica_before_started_at = {identity[1] for identity in replica_before_ids} + assert len(replica_before_ids) == len( + set(replica_before_ids) + ), f"duplicate replica hydration identities before restart: {replica_before}" + latest_before_finish = max(parse_ts(episode[2]) for episode in replica_before) + c.kill("materialized") c.up("materialized") @@ -1579,8 +1650,46 @@ def episodes(name: str = "hydration_history_i") -> list[list]: len(after) == 2 and len(fresh) == 1 ), f"expected one preserved and one fresh episode, got {after}" - # Let several sweeps run. The pre-restart episode must not be duplicated, - # and the post-restart episode must settle at one row too. + deadline = time.time() + 120 + replica_after = [] + replica_after_ids = [] + replica_fresh_ids = set() + while time.time() < deadline: + replica_after = replica_episodes() + replica_after_ids = replica_episode_identities(replica_after) + replica_fresh_ids = set(replica_after_ids) - set(replica_before_ids) + if set(replica_before_ids) <= set(replica_after_ids) and replica_fresh_ids: + break + time.sleep(0.5) + assert all( + episode in replica_after for episode in replica_before + ), f"restart changed replica episodes: had {replica_before}, now {replica_after}" + assert set(replica_before_ids) <= set( + replica_after_ids + ), f"restart lost replica episodes: had {replica_before}, now {replica_after}" + # Rehydration can record one fresh episode or several: the + # introspection indexes can finish before the user dataflows install, + # forming an earlier disconnected episode that is recorded on its own. + # The monotonic history guard orders them all after pre-restart + # history. + assert ( + replica_fresh_ids + ), f"restart did not produce a fresh replica identity: {replica_after}" + assert all( + parse_ts(identity[1]) > latest_before_finish + for identity in replica_fresh_ids + ), f"fresh replica episode overlaps pre-restart history: {replica_after}" + assert all( + identity[1] not in replica_before_started_at + for identity in replica_fresh_ids + ), f"restart reused a replica hydration start: {replica_after}" + assert len(replica_after_ids) == len( + set(replica_after_ids) + ), f"restart produced duplicate replica hydration identities: {replica_after}" + + # Let several sweeps run. The pre-restart episodes must not be + # duplicated, and everything recorded since the restart must stay + # ordered after them. time.sleep(10) settled = episodes() assert ( @@ -1591,6 +1700,25 @@ def episodes(name: str = "hydration_history_i") -> list[list]: ), f"sweeps duplicated a hydration episode: {settled}" assert len(settled) == 2, f"expected two settled episodes, got {settled}" + replica_settled = replica_episodes() + replica_settled_ids = replica_episode_identities(replica_settled) + assert len(replica_settled_ids) == len( + set(replica_settled_ids) + ), f"sweeps duplicated a replica hydration identity: {replica_settled}" + assert set(replica_before_ids) <= set( + replica_settled_ids + ), f"pre-restart replica episodes disappeared: {replica_settled}" + assert all( + episode in replica_settled for episode in replica_before + ), f"pre-restart replica episodes changed: {replica_settled}" + assert replica_fresh_ids <= set( + replica_settled_ids + ), f"post-restart replica episodes disappeared: {replica_settled}" + assert all( + parse_ts(identity[1]) > latest_before_finish + for identity in set(replica_settled_ids) - set(replica_before_ids) + ), f"settled replica episodes overlap pre-restart history: {replica_settled}" + def workflow_default(c: Composition) -> None: def process(name: str) -> None: diff --git a/test/sqllogictest/autogenerated/mz_internal.slt b/test/sqllogictest/autogenerated/mz_internal.slt index 47c55bb3d6813..e96388950a124 100644 --- a/test/sqllogictest/autogenerated/mz_internal.slt +++ b/test/sqllogictest/autogenerated/mz_internal.slt @@ -399,6 +399,18 @@ started_at timestamp␠with␠time␠zone When␠hydration␠work␠began,␠o hydrated_at timestamp␠with␠time␠zone When␠hydration␠finished. status text The␠terminal␠status.␠Currently␠always␠`hydrated`. +query TTT +SELECT name, type, comment FROM objects WHERE schema = 'mz_internal' AND object = 'mz_replica_hydration_history' ORDER BY position +---- +replica_id text The␠ID␠of␠the␠cluster␠replica.␠May␠name␠a␠replica␠that␠no␠longer␠exists. +cluster_id text The␠ID␠of␠the␠replica's␠cluster. +started_at timestamp␠with␠time␠zone The␠earliest␠maintained␠compute␠dataflow␠installation␠in␠the␠hydration␠episode. +finished_at timestamp␠with␠time␠zone The␠latest␠maintained␠compute␠dataflow␠hydration␠in␠the␠hydration␠episode. +object_count uint8 The␠number␠of␠maintained␠compute␠dataflows␠in␠the␠hydration␠episode. +peak_memory_bytes uint8 The␠largest␠process-lifetime␠cgroup␠memory␠high-water␠mark␠reported␠by␠any␠process␠when␠the␠collector␠recorded␠the␠episode.␠`NULL`␠if␠the␠platform␠reports␠no␠cgroup␠memory␠peak. +peak_disk_bytes uint8 The␠largest␠process-lifetime␠scratch-filesystem␠or␠swap␠high-water␠mark␠reported␠by␠any␠process␠when␠the␠collector␠recorded␠the␠episode.␠Filesystem␠peaks␠are␠sampled␠lower␠bounds.␠`NULL`␠if␠neither␠measurement␠is␠available. +status text The␠hydration␠episode's␠status.␠Currently␠always␠`hydrated`. + query TTT SELECT name, type, comment FROM objects WHERE schema = 'mz_internal' AND object = 'mz_object_transitive_dependencies' ORDER BY position ---- @@ -872,6 +884,7 @@ mz_recent_activity_log_thinned mz_recent_sql_text mz_recent_sql_text_redacted mz_replacements +mz_replica_hydration_history mz_replica_system_parameters mz_session_history mz_sessions diff --git a/test/sqllogictest/catalog_server_explain.slt b/test/sqllogictest/catalog_server_explain.slt index b06437fb06918..f28dde2030834 100644 --- a/test/sqllogictest/catalog_server_explain.slt +++ b/test/sqllogictest/catalog_server_explain.slt @@ -5472,7 +5472,7 @@ mz_catalog.mz_tables: Project: #3, #0, #4, #1, #5, #2, #6, #6, #6 Map: "s1", null →Arrange (#1{schema_name}, #2{name}) - →Constant (30 rows) + →Constant (31 rows) →Arrange (#0{schema_name}) (#0{schema_name}, #1{name}) →Fused with Child Map/Filter/Project Project: #5, #4, #6 @@ -8246,7 +8246,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_builtin_tables"; ---- Explained Query (fast path): - →Constant (30 rows) + →Constant (31 rows) Target cluster: mz_catalog_server @@ -10238,7 +10238,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_ontology_entity_types"; ---- Explained Query (fast path): - →Constant (137 rows) + →Constant (138 rows) Target cluster: mz_catalog_server @@ -10248,7 +10248,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_ontology_link_types"; ---- Explained Query (fast path): - →Constant (180 rows) + →Constant (182 rows) Target cluster: mz_catalog_server @@ -10262,7 +10262,7 @@ Explained Query: cte l0 = →Differential Join %1:mz_schemas[#0{id}] » %2:mz_objects[#2{schema_id}] » %0[#0{schema_name}, #1{table_name}] » %3:mz_columns[#0{id}] →Arrange (#0{schema_name}, #1{table_name}) - →Constant (137 rows) + →Constant (138 rows) →Arrange (#0{id}) →Fused with Child Map/Filter/Project Project: #1, #3 @@ -10315,7 +10315,7 @@ Explained Query: →Differential Join %0:l4[#0{entity_name}, #1{name}] » %1[#0{entity_name}, #1{column_name}] →Arranged l4 →Arrange (#0{entity_name}, #1{column_name}) - →Constant (281 rows) + →Constant (283 rows) →Return →Union →Map/Filter/Project diff --git a/test/sqllogictest/information_schema_tables.slt b/test/sqllogictest/information_schema_tables.slt index 4d0d5d7ac2a25..881b0b0b89c44 100644 --- a/test/sqllogictest/information_schema_tables.slt +++ b/test/sqllogictest/information_schema_tables.slt @@ -597,6 +597,10 @@ mz_replacements BASE TABLE materialize mz_internal +mz_replica_hydration_history +BASE TABLE +materialize +mz_internal mz_replica_system_parameters MATERIALIZED VIEW materialize diff --git a/test/sqllogictest/mz_catalog_server_index_accounting.slt b/test/sqllogictest/mz_catalog_server_index_accounting.slt index e8c638c152d60..a7da05083daf9 100644 --- a/test/sqllogictest/mz_catalog_server_index_accounting.slt +++ b/test/sqllogictest/mz_catalog_server_index_accounting.slt @@ -85,7 +85,7 @@ mz_message_batch_counts_sent_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_ba mz_message_counts_received_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_counts_received_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_counts_received_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id") mz_message_counts_sent_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_counts_sent_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_counts_sent_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id") mz_metric_sinks_ind CREATE␠INDEX␠"mz_metric_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s546␠AS␠"mz_internal"."mz_metric_sinks"]␠("id") -mz_notices_ind CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s843␠AS␠"mz_internal"."mz_notices"]␠("id") +mz_notices_ind CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s844␠AS␠"mz_internal"."mz_notices"]␠("id") mz_object_arrangement_size_history_object_ind CREATE␠INDEX␠"mz_object_arrangement_size_history_object_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s758␠AS␠"mz_internal"."mz_object_arrangement_size_history"]␠("object_id") mz_object_arrangement_size_history_ts_ind CREATE␠INDEX␠"mz_object_arrangement_size_history_ts_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s758␠AS␠"mz_internal"."mz_object_arrangement_size_history"]␠("collection_timestamp") mz_object_arrangement_sizes_ind CREATE␠INDEX␠"mz_object_arrangement_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s756␠AS␠"mz_internal"."mz_object_arrangement_sizes"]␠("replica_id") diff --git a/test/sqllogictest/oid.slt b/test/sqllogictest/oid.slt index ba84f1540219a..fe4fdaaebd174 100644 --- a/test/sqllogictest/oid.slt +++ b/test/sqllogictest/oid.slt @@ -1253,3 +1253,4 @@ SELECT oid, name FROM mz_objects WHERE id LIKE 's%' AND oid < 20000 ORDER BY oid 17121 mz_metric_sinks_ind 17122 mz_object_hydration_history 17123 mz_cluster_replica_resource_usage +17124 mz_replica_hydration_history diff --git a/test/testdrive/catalog.td b/test/testdrive/catalog.td index d783af6de9cce..12074488a6a79 100644 --- a/test/testdrive/catalog.td +++ b/test/testdrive/catalog.td @@ -613,6 +613,7 @@ mz_object_global_ids "" mz_object_hydration_history "" mz_optimizer_notices "" mz_replacements "" +mz_replica_hydration_history "" mz_sessions "" mz_source_references "" mz_storage_usage_by_shard "" @@ -836,7 +837,7 @@ test_table "" # `SHOW TABLES` and `mz_tables` should agree. > SELECT COUNT(*) FROM mz_tables WHERE id LIKE 's%' -30 +31 # There is one entry in mz_indexes for each field_number/expression of the index. > SELECT COUNT(id) FROM mz_indexes WHERE id LIKE 's%' diff --git a/test/testdrive/hydration-status.td b/test/testdrive/hydration-status.td index 2757c76d3dfb1..de043b67ff35e 100644 --- a/test/testdrive/hydration-status.td +++ b/test/testdrive/hydration-status.td @@ -31,8 +31,12 @@ ALTER SYSTEM SET hydration_history_collection_interval = '1s'; # Pin retention too: CI randomizes it, and a short period would prune the # episodes asserted below before the assertions run. ALTER SYSTEM SET hydration_history_retention_period = '30d'; +# Keep both publications of the process resource observations fresh enough to +# compare within the test's retry budget. +ALTER SYSTEM SET mz_metrics_usage_refresh_interval = '1s'; +ALTER SYSTEM SET compute_prometheus_introspection_scrape_interval = '1s'; -> CREATE CLUSTER test REPLICAS (hydrated_test_1 (SIZE 'scale=1,workers=1')) +> CREATE CLUSTER test REPLICAS (hydrated_test_1 (SIZE 'scale=2,workers=2')) > SET cluster = test # Test that on an empty cluster only the introspection indexes show up. @@ -57,6 +61,23 @@ si true JOIN mz_cluster_replicas r ON (r.id = h.replica_id) WHERE r.name LIKE 'hydrated_test%'; +# A system-only replica hydration episode covers every visible non-transient +# export. Its existence also proves the collector excludes its own transient +# subscribe, whose intervals would otherwise show up as spurious episodes in +# later snapshots. +> SELECT h.status, + h.object_count = ( + SELECT count(DISTINCT export_id)::uint8 + FROM mz_introspection.mz_compute_hydration_times_per_worker + WHERE export_id NOT LIKE 't%' + ) + FROM mz_internal.mz_replica_hydration_history AS h + JOIN mz_catalog.mz_cluster_replicas AS r ON r.id = h.replica_id + WHERE r.name = 'hydrated_test_1' + ORDER BY h.started_at DESC + LIMIT 1; +hydrated true + # Test adding new compute dataflows. > CREATE TABLE t (a int) @@ -87,11 +108,11 @@ mv_const hydrated_test_1 true false # turn cannot be recorded, rather than a special case. `idx` and `mv` keep their # dataflows installed, so they are stable. # -# The recorded interval must be ordered. The stamps are aggregated over a replica's -# workers, so on a multi-process replica the ends can come from different process -# clocks and the interval carries their skew. These replicas are single-process, so -# the ordering is exact here. The coalesce covers a NULL start, which the replica no -# longer produces but the column still permits. +# The recorded interval must be ordered. The stamps are aggregated over a +# replica's workers, so the ends can come from different process clocks and the +# duration carries their skew. Taking the earliest installation and latest +# hydration preserves ordering. The coalesce covers a NULL start, which the +# replica no longer produces but the column still permits. > SELECT o.name, r.name, h.status, h.installed_at <= h.hydrated_at, coalesce(h.installed_at <= h.started_at AND h.started_at <= h.hydrated_at, true) @@ -104,6 +125,133 @@ mv_const hydrated_test_1 true false idx hydrated_test_1 hydrated true true mv hydrated_test_1 hydrated true true +# Replica history rolls overlapping object intervals into one episode and waits +# for every configured process to publish resource observations. The exact +# number of initial episodes depends on whether these sequential DDL statements +# overlapped, so assert the stable properties here. +> SELECT count(*) > 0, + bool_and(h.status = 'hydrated' AND h.started_at <= h.finished_at), + bool_and(h.object_count > 0) + FROM mz_internal.mz_replica_hydration_history AS h + JOIN mz_catalog.mz_cluster_replicas AS r ON r.id = h.replica_id + WHERE r.name = 'hydrated_test_1'; +true true true + +# Installing an object after the replica is fully hydrated starts a new replica +# episode. Its boundaries match the same per-object aggregates used by object +# history. +> SET cluster_replica = hydrated_test_1 + +> SELECT count(*) > 0 + FROM mz_introspection.mz_cluster_prometheus_metrics + WHERE metric_name = 'mz_metrics_resource_usage'; +true + +$ set-from-sql var=memory-peak-before +SELECT coalesce(max(value)::text, '-1') +FROM mz_introspection.mz_cluster_replica_resource_usage +WHERE source = 'cgroup' + AND metric = 'memory_peak' + +$ set-from-sql var=disk-peak-before +SELECT coalesce(coalesce( + max(value) FILTER ( + WHERE source = 'statvfs' + AND metric = 'fs_used_peak' + ), + max(value) FILTER ( + WHERE source = 'cgroup' + AND metric = 'swap_peak' + ) +)::text, '-1') +FROM mz_introspection.mz_cluster_replica_resource_usage + +> RESET cluster_replica + +> CREATE INDEX replica_episode_idx ON t (a + 1) + +# This timeout bounds the complete production path: installing the targeted +# subscribe, evaluating the connected-component query, and committing through +# OCC. It is intentionally generous for debug builds but far below the +# collector's five-minute mutation timeout. Test configurations with a larger +# default can raise this bound for sanitizer overhead. +$ set-sql-timeout duration=20s + +> SELECT rh.status, rh.object_count, + rh.started_at = oh.installed_at, + rh.finished_at = oh.hydrated_at + FROM mz_internal.mz_replica_hydration_history AS rh + JOIN mz_internal.mz_object_hydration_history AS oh + ON oh.replica_id = rh.replica_id + AND oh.installed_at = rh.started_at + JOIN mz_internal.mz_object_global_ids AS g ON g.global_id = oh.object_id + JOIN mz_catalog.mz_objects AS o ON o.id = g.id + WHERE o.name = 'replica_episode_idx'; +hydrated 1 true true + +# The Prometheus registry and resource-usage log publish the same process +# sampler independently. A Prometheus scrape after collection can observe a +# higher process-lifetime peak, so the value is bracketed by the maximum across +# process resource logs before collection and the maximum across Prometheus +# series after it rather than required to equal the later scrape. Metric +# availability must agree. +> SET cluster_replica = hydrated_test_1 + +> WITH recorded AS ( + SELECT rh.peak_memory_bytes, rh.peak_disk_bytes + FROM mz_internal.mz_replica_hydration_history AS rh + JOIN mz_internal.mz_object_hydration_history AS oh + ON oh.replica_id = rh.replica_id + AND oh.installed_at = rh.started_at + JOIN mz_internal.mz_object_global_ids AS g ON g.global_id = oh.object_id + JOIN mz_catalog.mz_objects AS o ON o.id = g.id + WHERE o.name = 'replica_episode_idx' + ), prometheus AS ( + SELECT + count(DISTINCT process_id) AS process_count, + max(value) FILTER ( + WHERE labels -> 'source' = 'cgroup' + AND labels -> 'metric' = 'memory_peak' + ) AS peak_memory_bytes, + coalesce( + max(value) FILTER ( + WHERE labels -> 'source' = 'statvfs' + AND labels -> 'metric' = 'fs_used_peak' + ), + max(value) FILTER ( + WHERE labels -> 'source' = 'cgroup' + AND labels -> 'metric' = 'swap_peak' + ) + ) AS peak_disk_bytes + FROM mz_introspection.mz_cluster_prometheus_metrics + WHERE metric_name = 'mz_metrics_resource_usage' + ) + SELECT + p.process_count = 2, + CASE WHEN p.peak_memory_bytes IS NULL THEN + r.peak_memory_bytes IS NULL AND ${memory-peak-before}::double precision = -1 + ELSE + r.peak_memory_bytes IS NOT NULL + AND ${memory-peak-before}::double precision <= r.peak_memory_bytes::double precision + AND r.peak_memory_bytes::double precision <= p.peak_memory_bytes + END, + CASE WHEN p.peak_disk_bytes IS NULL THEN + r.peak_disk_bytes IS NULL AND ${disk-peak-before}::double precision = -1 + ELSE + r.peak_disk_bytes IS NOT NULL + AND ${disk-peak-before}::double precision <= r.peak_disk_bytes::double precision + AND r.peak_disk_bytes::double precision <= p.peak_disk_bytes + END + FROM recorded AS r + CROSS JOIN prometheus AS p; +true true true + +> RESET cluster_replica + +$ set-sql-timeout duration=default + +> DROP INDEX replica_episode_idx + > SELECT o.name, r.name, h.hydrated FROM mz_internal.mz_hydration_statuses h JOIN mz_cluster_replicas r ON (r.id = h.replica_id) @@ -409,6 +557,9 @@ ALTER SYSTEM SET hydration_history_retention_period = '0s'; > SELECT count(*) FROM mz_internal.mz_object_hydration_history; 0 +> SELECT count(*) FROM mz_internal.mz_replica_hydration_history; +0 + > SELECT o.name, r.name, h.hydrated FROM mz_internal.mz_hydration_statuses h JOIN mz_cluster_replicas r ON (r.id = h.replica_id) @@ -481,6 +632,121 @@ mv_wmr_stuck hydrated_test_4 false > DROP MATERIALIZED VIEW mv_wmr_const > DROP MATERIALIZED VIEW mv_wmr_stuck +# An in-progress replica episode must not block recording an earlier, +# disconnected completed episode. A gate that waits for every live export to +# hydrate is lossy, not just late: once the in-progress episode completes it is +# the latest, and the earlier episode is never recorded. Retention is still 0s +# here, so the history stays empty while the fixture is arranged and the +# assertion below sees exactly the episodes created now. + +> CREATE CLUSTER episodes REPLICAS (episodes_r1 (SIZE 'scale=1,workers=1')) +> SET cluster = episodes + +> CREATE INDEX episode_a_idx ON t (a) + +# Wait until every non-transient export on the replica has hydrated before +# installing the stuck view. A's interval can sit inside the introspection log +# indexes' interval (they hydrate concurrently), so only a fully hydrated +# replica guarantees that the stuck view's interval starts a new episode +# instead of connecting to the one containing A. +> SET cluster_replica = episodes_r1 + +> SELECT count(*) > 0 AND count(*) = count(h.hydrated_at), + count(*) FILTER (WHERE o.name = 'episode_a_idx') = 1 + FROM mz_introspection.mz_compute_hydration_times_per_worker AS h + LEFT JOIN mz_internal.mz_object_global_ids AS g ON g.global_id = h.export_id + LEFT JOIN mz_catalog.mz_objects AS o ON o.id = g.id + WHERE h.export_id NOT LIKE 't%'; +true true + +# The finish of the episode containing A: the latest hydration on the fully +# hydrated replica. Stamps are immutable, and nothing else is installed before +# the stuck view, so this is the finish the collector must record. +$ set-from-sql var=episode-finished-at +SELECT max(hydrated_at)::text +FROM mz_introspection.mz_compute_hydration_times_per_worker +WHERE export_id NOT LIKE 't%' + +> RESET cluster_replica + +# Whether this view is stuck or merely slow is not observable from the +# hydration log, so its open interval is an in-progress episode, not an error. +> CREATE MATERIALIZED VIEW episode_stuck AS + WITH MUTUALLY RECURSIVE + x (a int) AS ( + VALUES (1) + UNION ALL + SELECT a + 1 FROM x + ) + SELECT * FROM x; + +> SET cluster_replica = episodes_r1 + +> SELECT count(*) > 0 AND count(h.hydrated_at) = 0 + FROM mz_introspection.mz_compute_hydration_times_per_worker AS h + JOIN mz_internal.mz_object_global_ids AS g ON g.global_id = h.export_id + JOIN mz_catalog.mz_objects AS o ON o.id = g.id + WHERE o.name = 'episode_stuck'; +true + +> RESET cluster_replica + +> CREATE INDEX episode_b_idx ON t (a + 1) + +# B hydrates while the stuck view's interval is open, which connects B to the +# in-progress episode: the episode containing A finished before the stuck view +# was installed, B after. +> SET cluster_replica = episodes_r1 + +> WITH + u AS ( + SELECT min(h.installed_at) AS installed_at + FROM mz_introspection.mz_compute_hydration_times_per_worker AS h + JOIN mz_internal.mz_object_global_ids AS g ON g.global_id = h.export_id + JOIN mz_catalog.mz_objects AS o ON o.id = g.id + WHERE o.name = 'episode_stuck' + ), + b AS ( + SELECT max(h.hydrated_at) AS finished_at, + count(*) > 0 AND count(*) = count(h.hydrated_at) AS hydrated + FROM mz_introspection.mz_compute_hydration_times_per_worker AS h + JOIN mz_internal.mz_object_global_ids AS g ON g.global_id = h.export_id + JOIN mz_catalog.mz_objects AS o ON o.id = g.id + WHERE o.name = 'episode_b_idx' + ) + SELECT TIMESTAMPTZ '${episode-finished-at}' < u.installed_at, + u.installed_at <= b.finished_at, + b.hydrated + FROM u, b; +true true true + +$ set-from-sql var=stuck-installed-at +SELECT min(h.installed_at)::text +FROM mz_introspection.mz_compute_hydration_times_per_worker AS h +JOIN mz_internal.mz_object_global_ids AS g ON g.global_id = h.export_id +JOIN mz_catalog.mz_objects AS o ON o.id = g.id +WHERE o.name = 'episode_stuck' + +> RESET cluster_replica + +# Re-enabling retention lets episodes be recorded again. Exactly A's episode +# lands: it is the latest completed episode disconnected from the in-progress +# one, which is not recorded (it never completes here) and carries B with it. +$ postgres-execute connection=mz_system +ALTER SYSTEM SET hydration_history_retention_period = '30d'; + +> SELECT count(*), + bool_and(h.status = 'hydrated'), + bool_and(h.finished_at = TIMESTAMPTZ '${episode-finished-at}'), + bool_and(h.finished_at < TIMESTAMPTZ '${stuck-installed-at}') + FROM mz_internal.mz_replica_hydration_history AS h + JOIN mz_catalog.mz_cluster_replicas AS r ON r.id = h.replica_id + WHERE r.name = 'episodes_r1'; +1 true true true + +> DROP CLUSTER episodes CASCADE +> SET cluster = test + # Test that incorrectly configured sinks do _not_ show as hydrated. > CREATE TABLE schema1 (a int) diff --git a/test/workload-replay/objects.txt b/test/workload-replay/objects.txt index ca116ead25529..d771dfca6654d 100644 --- a/test/workload-replay/objects.txt +++ b/test/workload-replay/objects.txt @@ -657,6 +657,7 @@ mz_records_per_dataflow_per_worker mz_relations mz_render_typmod mz_replacements +mz_replica_hydration_history mz_resolve_object_name mz_role_auth mz_role_members diff --git a/test/workload-replay/system_catalog_identifiers.txt b/test/workload-replay/system_catalog_identifiers.txt index 354f1df0e5662..ca71211a6f844 100644 --- a/test/workload-replay/system_catalog_identifiers.txt +++ b/test/workload-replay/system_catalog_identifiers.txt @@ -977,6 +977,7 @@ mz_records_per_dataflow_per_worker mz_relations mz_render_typmod mz_replacements +mz_replica_hydration_history mz_replica_system_parameters mz_resolve_object_name mz_role_auth