From 37aeca424452e214d9154471c635f29f4afed443 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?G=C3=A1bor=20E=2E=20G=C3=A9vay?= Date: Wed, 2 Sep 2026 09:43:02 +0200 Subject: [PATCH 1/3] sql-parser: handle EXPLAIN TIMESTAMP AS DOT instead of panicking (#38601) ### Motivation `EXPLAIN TIMESTAMP AS DOT FOR ` panics the parser (`unreachable!()` at `src/sql-parser/src/parser.rs`), and since the parse path is not wrapped in `catch_unwind` and the panic hook calls `process::abort()`, it crashes environmentd. Post-authentication, so any role that can run SQL can take the environment down with one statement. Verified on the v26.39.0 emulator (SIGABRT, exit 134). Found while reviewing the `EXPLAIN TIMESTAMP` docs page. Closes: [SQL-665](https://linear.app/materializeinc/issue/SQL-665) ### Description `parse_explain_timestamp` accepts `DOT` in `parse_one_of_keywords(&[TEXT, JSON, DOT])` but had no `Some(DOT)` arm, so `AS DOT` fell through to `_ => unreachable!()`. Added `Some(DOT) => Some(ExplainFormat::Dot)`, matching `parse_explain_plan`. The statement now parses and reaches the sequencer's existing `AdapterError::Unsupported("EXPLAIN TIMESTAMP AS DOT")` (`src/adapter/src/coord/sequencer/inner/explain_timestamp.rs`) instead of panicking. ### Verification Added a `parse-statement` regression case for `EXPLAIN TIMESTAMP AS DOT FOR SELECT 1` to `src/sql-parser/tests/testdata/explain`. Co-authored-by: Claude Fable 5 --- src/sql-parser/src/parser.rs | 1 + src/sql-parser/tests/testdata/explain | 7 +++++++ 2 files changed, 8 insertions(+) diff --git a/src/sql-parser/src/parser.rs b/src/sql-parser/src/parser.rs index d878a936530da..3f82881d5ca9c 100644 --- a/src/sql-parser/src/parser.rs +++ b/src/sql-parser/src/parser.rs @@ -9478,6 +9478,7 @@ impl<'a> Parser<'a> { match self.parse_one_of_keywords(&[TEXT, JSON, DOT]) { Some(TEXT) => Some(ExplainFormat::Text), Some(JSON) => Some(ExplainFormat::Json), + Some(DOT) => Some(ExplainFormat::Dot), None => return Err(ParserError::new(self.index, "expected a format")), _ => unreachable!(), } diff --git a/src/sql-parser/tests/testdata/explain b/src/sql-parser/tests/testdata/explain index 17bc338d82b50..103dcd4294985 100644 --- a/src/sql-parser/tests/testdata/explain +++ b/src/sql-parser/tests/testdata/explain @@ -194,6 +194,13 @@ EXPLAIN TIMESTAMP AS JSON FOR SELECT 1 => ExplainTimestamp(ExplainTimestampStatement { format: Some(Json), select: SelectStatement { query: Query { ctes: Simple([]), body: Select(Select { distinct: None, projection: [Expr { expr: Value(Number("1")), alias: None }], from: [], selection: None, group_by: [], having: None, qualify: None, options: [] }), order_by: [], limit: None, offset: None }, as_of: None } }) +parse-statement +EXPLAIN TIMESTAMP AS DOT FOR SELECT 1 +---- +EXPLAIN TIMESTAMP AS DOT FOR SELECT 1 +=> +ExplainTimestamp(ExplainTimestampStatement { format: Some(Dot), select: SelectStatement { query: Query { ctes: Simple([]), body: Select(Select { distinct: None, projection: [Expr { expr: Value(Number("1")), alias: None }], from: [], selection: None, group_by: [], having: None, qualify: None, options: [] }), order_by: [], limit: None, offset: None }, as_of: None } }) + parse-statement EXPLAIN AS JSON SELECT * FROM foo ---- From 4e012ea88d0d5d4b03d42dc43495e2eb37ba6b7b Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Wed, 2 Sep 2026 11:00:26 +0200 Subject: [PATCH 2/3] Fix clippy warnings on newer Rust toolchains (#38619) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI derives its toolchain from the `rust-version` field in the root `Cargo.toml`, which pins 1.97.1. Warnings that newer toolchains introduce therefore go unreported until someone builds locally. This change clears everything that Rust 1.98 reports, plus the deprecations that the current nightly adds on top. Rust 1.98 extends `clippy::extra_unused_lifetimes` to higher-ranked binders in where-clauses, which flags three `for<'a>` binders whose lifetime never appears in the bound they introduce. Removing the binder leaves the bounds unchanged, because the parenthesized `Fn` sugar already introduces a fresh lifetime for each elided reference. The same toolchain also reports two unused imports and three `clippy::useless_format` calls. `parser_err!` calls `to_string` on a single-expression argument, so passing the literal directly still produces an owned `String`. Nightly additionally deprecates the legacy numeric constants and constructors. `max_value` and `min_value` become the `MAX` and `MIN` associated constants, and `std::usize::MAX` becomes `usize::MAX`. Dropping the `use std::i64` import in the Avro decoder is what silences the two warnings on `i64::MAX` and `i64::MIN`, because that import made the path resolve to the deprecated module instead of the primitive type. Two groups of nightly warnings are left alone. The `fetch_update` to `try_update` rename is too recent to use while CI builds with 1.97.1, and the `overflow evaluating the requirement` notices come from the new trait solver hitting its recursion limit rather than from a defect in this code. Note that this change does not close the gap that hid these warnings. CI keeps building with 1.97.1, so the next lint a newer toolchain adds will go unreported the same way. Bumping `rust-version` rebuilds the CI builder image and raises the declared minimum supported version, which belongs in its own change. ### Release notes No user-visible changes. 🤖 Posted by [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5 (1M context) --- src/adapter/src/catalog/open/builtin_schema_migration.rs | 2 +- src/avro/src/util.rs | 3 +-- src/ccsr/tests/client.rs | 2 +- src/compute/src/logging.rs | 2 +- src/compute/src/typedefs.rs | 2 +- src/expr/src/linear.rs | 2 +- src/expr/src/scalar/func.rs | 1 - src/orchestrator/src/lib.rs | 2 +- src/sql-parser/src/parser.rs | 6 +++--- src/timely-util/src/columnar/merge_batcher.rs | 1 - 10 files changed, 10 insertions(+), 13 deletions(-) diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index 5ddc4839943e5..e34fc1b1f4ffa 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -1400,7 +1400,7 @@ async fn read_migration_shard

( predicate: P, ) -> Option> where - P: for<'a> Fn(&migration_shard::Key) -> bool, + P: Fn(&migration_shard::Key) -> bool, { let as_of = Antichain::from_elem(read_ts); let updates = persist_read.snapshot_and_fetch(as_of).await.ok()?; diff --git a/src/avro/src/util.rs b/src/avro/src/util.rs index 2d8fbff99e6d9..610b584123b81 100644 --- a/src/avro/src/util.rs +++ b/src/avro/src/util.rs @@ -21,7 +21,6 @@ // The original source code is subject to the terms of the MIT license, a copy // of which can be found in the LICENSE file at the root of this repository. -use std::i64; use std::io::Read; use serde_json::{Map, Value}; @@ -73,7 +72,7 @@ pub fn zig_i64(n: i64, buffer: &mut Vec) { pub fn zag_i32(reader: &mut R) -> Result { let i = zag_i64(reader)?; - if i < i64::from(i32::min_value()) || i > i64::from(i32::max_value()) { + if i < i64::from(i32::MIN) || i > i64::from(i32::MAX) { Err(AvroError::Decode(DecodeError::I32OutOfRange(i))) } else { Ok(i as i32) diff --git a/src/ccsr/tests/client.rs b/src/ccsr/tests/client.rs index 364dbb5ef67a9..17798cc30dcbc 100644 --- a/src/ccsr/tests/client.rs +++ b/src/ccsr/tests/client.rs @@ -268,7 +268,7 @@ async fn test_client_errors() -> Result<(), anyhow::Error> { let client = mz_ccsr::ClientConfig::new(SCHEMA_REGISTRY_URL.clone()).build()?; // Get-by-id-specific errors. - match client.get_schema_by_id(i32::max_value()).await { + match client.get_schema_by_id(i32::MAX).await { Err(GetByIdError::SchemaNotFound) => (), res => panic!("expected GetError::SchemaNotFound, got {:?}", res), } diff --git a/src/compute/src/logging.rs b/src/compute/src/logging.rs index e8352b134ca9a..c27228d84f0d8 100644 --- a/src/compute/src/logging.rs +++ b/src/compute/src/logging.rs @@ -306,7 +306,7 @@ where B::Output: Clone, CB: ContainerBuilder, L: Into, - F: for<'a> FnMut(B::Output, &mut PermutedRowPacker, &mut OutputSession) + 'static, + F: FnMut(B::Output, &mut PermutedRowPacker, &mut OutputSession) + 'static, { let log = log.into(); // TODO: Use something other than the debug representation of the log variant as a name. diff --git a/src/compute/src/typedefs.rs b/src/compute/src/typedefs.rs index 8414445e1c945..0c71a624f5fbb 100644 --- a/src/compute/src/typedefs.rs +++ b/src/compute/src/typedefs.rs @@ -145,7 +145,7 @@ pub trait MzData: impl MzData for T where T: columnation::Columnation, - T: for<'a> columnar::Columnar, + T: columnar::Columnar, for<'a> Ref<'a, T>: Copy + Ord, { } diff --git a/src/expr/src/linear.rs b/src/expr/src/linear.rs index 2b2c0cb692b2b..8725a7b91aec4 100644 --- a/src/expr/src/linear.rs +++ b/src/expr/src/linear.rs @@ -977,7 +977,7 @@ impl MapFilterProject { pub fn optimize(&mut self) { // Track sizes and iterate as long as they decrease. let mut prev_size = None; - let mut self_size = usize::max_value(); + let mut self_size = usize::MAX; // Continue as long as strict improvements occur. while prev_size.map(|p| self_size < p).unwrap_or(true) { // Lock in current size. diff --git a/src/expr/src/scalar/func.rs b/src/expr/src/scalar/func.rs index 40b55d2c4b75d..fda3137ef95dc 100644 --- a/src/expr/src/scalar/func.rs +++ b/src/expr/src/scalar/func.rs @@ -3295,7 +3295,6 @@ fn mz_render_typmod(oid: u32, typmod: i32) -> String { #[cfg(test)] mod test { - use chrono::prelude::*; use mz_repr::PropDatum; use proptest::prelude::*; diff --git a/src/orchestrator/src/lib.rs b/src/orchestrator/src/lib.rs index b8c2e90824a53..5503065867aa1 100644 --- a/src/orchestrator/src/lib.rs +++ b/src/orchestrator/src/lib.rs @@ -388,7 +388,7 @@ impl<'de> Deserialize<'de> for CpuLimit { // Note -- we just round off any precision beyond 0.001 here. let float = f64::deserialize(deserializer)?; let millicpus = (float * 1000.).round(); - if millicpus < 0. || millicpus > (std::usize::MAX as f64) { + if millicpus < 0. || millicpus > (usize::MAX as f64) { use serde::de::Error; Err(D::Error::invalid_value( Unexpected::Float(float), diff --git a/src/sql-parser/src/parser.rs b/src/sql-parser/src/parser.rs index 3f82881d5ca9c..89df3729a5408 100644 --- a/src/sql-parser/src/parser.rs +++ b/src/sql-parser/src/parser.rs @@ -9539,7 +9539,7 @@ impl<'a> Parser<'a> { let err = parser_err!( self, self.peek_prev_pos(), - format!("WITH HOLD is unsupported for cursors") + "WITH HOLD is unsupported for cursors" ) .map_parser_err(StatementKind::Declare); self.expect_keyword(HOLD) @@ -10255,7 +10255,7 @@ impl<'a> Parser<'a> { return parser_err!( self, self.peek_prev_pos(), - format!("For object type MATERIALIZED VIEWS, you must specify 'TABLES'") + "For object type MATERIALIZED VIEWS, you must specify 'TABLES'" ); } @@ -10298,7 +10298,7 @@ impl<'a> Parser<'a> { return parser_err!( self, self.peek_prev_pos(), - format!("For object type MATERIALIZED VIEWS, you must specify 'TABLES'") + "For object type MATERIALIZED VIEWS, you must specify 'TABLES'" ); } diff --git a/src/timely-util/src/columnar/merge_batcher.rs b/src/timely-util/src/columnar/merge_batcher.rs index 8135933323430..05a64860a1462 100644 --- a/src/timely-util/src/columnar/merge_batcher.rs +++ b/src/timely-util/src/columnar/merge_batcher.rs @@ -667,7 +667,6 @@ mod tests { use std::sync::Arc; use columnar::Index; - use timely::container::PushInto as _; use super::*; use crate::column_pager::{PageDecision, PageEvent, PageHint, PagingPolicy}; From 46f729653aeb29ba77544ab3b2a3f7bff0a2f676 Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Wed, 2 Sep 2026 11:17:52 +0200 Subject: [PATCH 3/3] adapter: collect durable replica hydration history (#38512) ### Motivation Materialize records successful hydration for individual indexes and materialized views, but it does not retain a replica-level history. This makes it difficult to answer when a replica finished hydrating, how many maintained compute exports participated, or what process resource high-water marks were visible when the episode was recorded. ### Description This change adds `mz_internal.mz_replica_hydration_history` and extends the existing durable hydration-history sweep to populate it. A replica hydration episode is a connected set of overlapping export intervals. 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. A gap between intervals means that the replica was fully hydrated before the next export was installed, so the next interval starts a new episode. 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 recorded when it completes and does not block earlier, disconnected completed episodes. Transient exports are excluded because the collection query creates one itself and its intervals would otherwise show up as spurious episodes in later snapshots. The object-level collector remains limited to user indexes and materialized views. Each row records replica and cluster IDs, episode boundaries, the number of compute exports in the episode, status, and the largest resource high-water marks reported by any replica process. Memory uses cgroup `memory_peak`. Disk uses sampled `statvfs` `fs_used_peak` when a scratch filesystem is present, with cgroup `swap_peak` as the fallback. Managed resource limits apply independently to each process, so the maximum across processes is relevant. Summing process peaks could combine values that did not occur simultaneously. Resource observations have process-lifetime scope through collection. They are associated with an episode but are not bounded by its timestamps. Work after hydration can raise them, and a later episode can retain an earlier process peak. Collection is snapshot sampling rather than an event log. Each sweep records only the latest completed episode visible in its snapshot. Intermediate episodes can be missed, and exports retracted before collection leave no evidence. Only successful episodes are currently recorded. Process-local clock skew can merge intervals that did not overlap in real time. The collector reuses the replica-targeted timestamped read-then-write path. Exact-timestamp OCC and an anti-join make concurrent collection idempotent. A monotonic history guard prevents a later snapshot from recording an episode that precedes or overlaps retained history, and still admits an in-progress episode once it completes, because its start necessarily lies after every recorded finish. Retention is bounded and eventually consistent across concurrent collectors. The table survives normal `environmentd` and replica restarts and is protected from forced replacement migrations. Collection remains disabled by default through `hydration_history_collection_interval`. Coordinator shutdown explicitly aborts and waits for an active sweep before releasing timestamp-oracle workers, which preserves dependency ordering for timestamp-oracle clients cached by the sweep. ### Tests The hydration testdrive coverage exercises the production collector. It verifies that a system-only replica episode includes every visible non-transient export while excluding the collector's transient subscribe. It also checks single-export episode boundaries, compares persisted memory and disk values with independently published resource metrics, and verifies that a completed episode is recorded while a later, disconnected export is still hydrating. The restart workflow verifies that retained rows survive `environmentd` and replica restarts, restarts record fresh episode identities ordered after all pre-restart history, and repeated sweeps do not duplicate an identity. It re-reads the pre-restart episode set immediately before the restart, so its assertions do not depend on collection timing. Unit coverage checks episode query construction, latest-episode selection, process-count gating, replica rotation, task cancellation, and migration policy. Catalog expectations cover the relation, comments, OID, ontology metadata, and system catalog plans. ### Release note This release will add `mz_internal.mz_replica_hydration_history`, an internal durable history of successful replica hydration episodes and their observed process resource high-water marks. --- ...60827_durable_replica_hydration_history.md | 82 ++++ .../reference/system-catalog/mz_internal.md | 25 ++ doc/user/data/metrics.yml | 2 +- src/adapter-types/src/dyncfgs.rs | 4 +- .../catalog/open/builtin_schema_migration.rs | 21 +- .../open/builtin_schema_migration_tests.rs | 25 +- src/adapter/src/coord.rs | 12 +- src/adapter/src/coord/hydration_history.rs | 384 ++++++++++++++++-- src/adapter/src/metrics.rs | 2 +- src/catalog/src/builtin.rs | 9 + src/catalog/src/builtin/mz_internal.rs | 108 +++++ src/ore/src/task.rs | 43 +- src/pgrepr-consts/src/oid.rs | 1 + test/restart/mzcompose.py | 164 +++++++- .../autogenerated/mz_internal.slt | 13 + test/sqllogictest/catalog_server_explain.slt | 12 +- .../information_schema_tables.slt | 4 + .../mz_catalog_server_index_accounting.slt | 2 +- test/sqllogictest/oid.slt | 1 + test/testdrive/catalog.td | 3 +- test/testdrive/hydration-status.td | 278 ++++++++++++- test/workload-replay/objects.txt | 1 + .../system_catalog_identifiers.txt | 1 + 23 files changed, 1103 insertions(+), 94 deletions(-) create mode 100644 doc/developer/design/20260827_durable_replica_hydration_history.md 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 c64ebccb5a33e..76d7e78b76ae4 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 e34fc1b1f4ffa..1754ccbf26250 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, }; @@ -698,18 +699,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, @@ -828,6 +831,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 bdc4ab5196caa..9d17f66ac1eba 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; @@ -3234,6 +3234,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(); @@ -4388,6 +4390,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 6fb54b7867f51..e70d02816250f 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 139aca6652e56..e8e154850df90 100644 --- a/src/catalog/src/builtin/mz_internal.rs +++ b/src/catalog/src/builtin/mz_internal.rs @@ -5248,6 +5248,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 ab3e3e9da686d..62a3ee04f6785 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 40435e9417dbe..54e9301255e67 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 c7c1fe1ac4bc1..05c34d29a228c 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 02d57e04b7507..1c95935dfd8e9 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