From ea704c4ddd7b375151abfd922f2af77e7f39a711 Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Fri, 25 Sep 2026 12:42:09 +0000 Subject: [PATCH] adapter: Compare 0dt hydration against the leader The incoming generation should not need more hydration than the leader has. Read the leader's mz_compute_hydration_times and waive local hydration where the leader explicitly reports none: * A compute collection within the lag bound need not hydrate when every leader replica hosting it reports it unhydrated. The stability gate then ignores it too. * A replica the leader reports unhydrated, for a collection the classification did not ignore, only has to be Online. A replica the leader has fully hydrated still has to hydrate here. Missing reports never waive anything. Protect the hydration source from builtin migration, so the check keeps reading the leader's shard. Closes: SQL-696 Co-authored-by: Amp Amp-Thread-ID: https://ampcode.com/threads/T-01a0d85f-8cd1-7078-afa3-21a9cce44857 --- .../catalog/open/builtin_schema_migration.rs | 14 +- src/adapter/src/coord/caught_up.rs | 225 ++++++++++++++++-- src/catalog/src/builtin.rs | 9 + test/0dt/mock_hydration_flags.py | 101 ++++++++ test/0dt/mzcompose.py | 179 ++++++++++++++ 5 files changed, 508 insertions(+), 20 deletions(-) create mode 100644 test/0dt/mock_hydration_flags.py diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index bbdf4ad2ab616..c8ed6bcebc5d0 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -37,11 +37,11 @@ use futures::future::BoxFuture; use mz_build_info::{BuildInfo, DUMMY_BUILD_INFO}; 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, + MZ_CLUSTER_REPLICA_FRONTIERS_DESCRIPTION, MZ_COMPUTE_HYDRATION_TIMES_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, }; use mz_catalog::config::BuiltinItemMigrationConfig; use mz_catalog::durable::objects::SystemObjectUniqueIdentifier; @@ -907,6 +907,10 @@ impl Migration { &*MZ_CLUSTER_REPLICA_FRONTIERS_DESCRIPTION, object, "mz_cluster_replica_frontiers cannot be migrated or else the 0dt caught-up gate loses its live-frontier reference" ); + assert_ne!( + &*MZ_COMPUTE_HYDRATION_TIMES_DESCRIPTION, object, + "mz_compute_hydration_times cannot be migrated or else the 0dt caught-up gate loses its leader-hydration reference" + ); let Some(object_info) = self.system_objects.get(object) else { panic!("migration step for non-existent builtin: {object:?}"); diff --git a/src/adapter/src/coord/caught_up.rs b/src/adapter/src/coord/caught_up.rs index 6681d5a46f6e5..c166a6a397a31 100644 --- a/src/adapter/src/coord/caught_up.rs +++ b/src/adapter/src/coord/caught_up.rs @@ -26,8 +26,9 @@ //! OOM-looping replica can momentarily look hydrated and caught-up, and cutting //! over right then drops us straight into a crashing replica. On top of the //! per-tick caught-up classification we therefore run a stability gate. A -//! caught-up cluster is ready only once every replica is `Online` and has had -//! everything it hosts hydrated for a configurable period. +//! caught-up cluster is ready only once every replica is `Online` and every +//! non-exempt replica has had its non-ignored collections hydrated for a +//! configurable period. //! //! Each replica reports its hydration times in its //! `mz_compute_hydration_times_per_worker` introspection log. The log lives in @@ -44,7 +45,8 @@ //! hosts keeping their clocks in sync. //! //! A replica without introspection logging cannot report hydration times. It -//! only has to be `Online`. +//! only has to be `Online`. Leader-relative exemptions are defined by +//! [`LeaderHydration::replica_has_unhydrated_collection`]. use std::collections::{BTreeMap, BTreeSet}; use std::fmt::Write as _; @@ -59,7 +61,9 @@ use mz_adapter_types::dyncfgs::{ WITH_0DT_CAUGHT_UP_CHECK_ALLOWED_LAG, WITH_0DT_CAUGHT_UP_CHECK_CUTOFF, WITH_0DT_CAUGHT_UP_CHECK_STABILITY_PERIOD, WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL, }; -use mz_catalog::builtin::{MZ_CLUSTER_REPLICA_FRONTIERS, MZ_CLUSTER_REPLICA_STATUS_HISTORY}; +use mz_catalog::builtin::{ + MZ_CLUSTER_REPLICA_FRONTIERS, MZ_CLUSTER_REPLICA_STATUS_HISTORY, MZ_COMPUTE_HYDRATION_TIMES, +}; use mz_catalog::memory::objects::Cluster; use mz_compute_client::controller::CollectionReadiness; use mz_compute_client::logging::{ComputeLog, LogVariant}; @@ -190,8 +194,7 @@ enum ReplicaHealth { NotHydrated, /// The replica has been healthy since this wall-clock time. HealthySince(EpochMillis), - /// The replica is `Online` but cannot report hydration times, so there is - /// nothing to wait on. + /// The replica is `Online` and exempt from the hydration stability check. Exempt, } @@ -489,9 +492,27 @@ impl Coordinator { }) .collect_vec(); - // The classification only requires each collection to be hydrated on - // _some_ replica. The stability gate separately requires every replica - // to have hydrated everything it hosts. + let leader_hydration = { + let item_id = self + .catalog() + .resolve_builtin_storage_collection(&MZ_COMPUTE_HYDRATION_TIMES); + let id = self.catalog().get_entry(&item_id).latest_global_id(); + match self + .controller + .storage_collections + .snapshot_latest(id) + .await + { + Ok(rows) => LeaderHydration::from_rows(rows), + Err(error) => { + tracing::warn!(%error, "cannot read leader hydration; requiring local hydration"); + LeaderHydration::default() + } + } + }; + let leader_unhydrated_collections = + leader_hydration.collections_unhydrated_on_all_hosting_replicas(&live_frontiers); + let live_collection_frontiers: BTreeMap<_, _> = live_frontiers .into_iter() .map(|(oid, _replica_id, upper_ts)| (oid, upper_ts)) @@ -548,6 +569,7 @@ impl Coordinator { cutoff.into(), now.into(), &live_collection_frontiers, + &leader_unhydrated_collections, &exclude_collections, &problematic_replicas, ) @@ -565,8 +587,12 @@ impl Coordinator { } => Some(( cluster_id, CaughtUpCluster { + replicas: self.replica_targets( + cluster_id, + &leader_hydration, + &ignored_compute_collections, + ), ignored_compute_collections, - replicas: self.replica_targets(cluster_id), }, )), _ => None, @@ -588,7 +614,12 @@ impl Coordinator { } /// Classifies the replicas of a cluster for the stability gate. - fn replica_targets(&self, cluster_id: ClusterId) -> BTreeMap { + fn replica_targets( + &self, + cluster_id: ClusterId, + leader: &LeaderHydration, + ignored: &BTreeSet, + ) -> BTreeMap { let cluster = self.catalog().get_cluster(cluster_id); let has_hydration_log = cluster .log_indexes @@ -607,7 +638,9 @@ impl Coordinator { let logging = has_hydration_log && replica.config.compute.logging.enabled(); let target = if !online { ReplicaTarget::NotOnline - } else if !logging { + } else if !logging + || leader.replica_has_unhydrated_collection(replica.replica_id, ignored) + { ReplicaTarget::Exempt } else { ReplicaTarget::Ask @@ -624,7 +657,9 @@ impl Coordinator { /// /// (1) A cluster is caught-up if all non-transient, non-excluded collections installed on it /// are either caught-up or ignored. - /// (2) A collection is caught-up when it is (a) hydrated and (b) its write frontier is within + /// (2) A collection is caught-up when it is (a) hydrated (waived for compute collections + /// returned by [`LeaderHydration::collections_unhydrated_on_all_hosting_replicas`]), + /// and (b) its write frontier is within /// `allowed_lag` of the "live" frontier, the collection's frontier reported by the leader /// environment. /// (3) A collection is ignored if its "live" frontier is behind `now` by more than `cutoff`. @@ -643,6 +678,7 @@ impl Coordinator { cutoff: Timestamp, now: Timestamp, live_frontiers: &BTreeMap>, + leader_unhydrated_collections: &BTreeSet, exclude_collections: &BTreeSet, problematic_replicas: &BTreeSet, ) -> BTreeMap { @@ -655,6 +691,7 @@ impl Coordinator { cutoff.clone(), now.clone(), live_frontiers, + leader_unhydrated_collections, exclude_collections, problematic_replicas, ) @@ -688,6 +725,7 @@ impl Coordinator { cutoff: Timestamp, now: Timestamp, live_frontiers: &BTreeMap>, + leader_unhydrated_collections: &BTreeSet, exclude_collections: &BTreeSet, problematic_replicas: &BTreeSet, ) -> Result { @@ -851,8 +889,10 @@ impl Coordinator { CollectionType::Storage => self.controller.storage.collection_hydrated(id)?, }; + let unhydrated_on_leader = matches!(collection_type, CollectionType::Compute) + && leader_unhydrated_collections.contains(&id); let readiness = CollectionReadiness::classify( - collection_hydrated, + collection_hydrated || unhydrated_on_leader, &write_frontier, Some((live_write_frontier, allowed_lag)), ); @@ -860,7 +900,8 @@ impl Coordinator { // We don't expect collections to get hydrated, ingestions to be // started, etc. when they are already at the empty write frontier. if live_write_frontier.is_empty() || readiness == CollectionReadiness::Ready { - if readiness != CollectionReadiness::Ready + // Carry hydration waivers into the per-replica stability check. + if (readiness != CollectionReadiness::Ready || !collection_hydrated) && matches!(collection_type, CollectionType::Compute) { ignored_compute_collections.insert(id); @@ -1048,6 +1089,83 @@ impl Coordinator { } } +/// The leader's hydration reports, read from its `mz_compute_hydration_times`. +#[derive(Debug, Default)] +struct LeaderHydration { + hydrated_on_any_replica: BTreeSet, + /// (Collection ID, replica ID) pairs explicitly reported as unhydrated. + unhydrated_replica_collections: BTreeSet<(GlobalId, String)>, +} + +impl LeaderHydration { + fn from_rows(rows: Vec) -> Self { + let mut leader = Self::default(); + for row in rows { + let mut values = row.iter(); + let replica = values + .next() + .expect("missing replica_id") + .unwrap_str() + .to_owned(); + let id: GlobalId = values + .next() + .expect("missing object_id") + .unwrap_str() + .parse() + .expect("valid object ID"); + if values.next().expect("missing time_ns").is_null() { + leader.unhydrated_replica_collections.insert((id, replica)); + } else { + leader.hydrated_on_any_replica.insert(id); + } + } + leader + } + + /// Returns the collections that every frontier-hosting leader replica + /// explicitly reports unhydrated. + /// + /// Missing reports or frontiers grant no waiver. A hydrated report from any + /// replica vetoes the waiver, even if that replica has no frontier report. + fn collections_unhydrated_on_all_hosting_replicas( + &self, + frontiers: &[(GlobalId, String, Antichain)], + ) -> BTreeSet { + let mut collections = BTreeMap::new(); + for (id, replica, _) in frontiers { + let all_unhydrated = collections.entry(*id).or_insert(true); + *all_unhydrated &= !self.hydrated_on_any_replica.contains(id) + && self + .unhydrated_replica_collections + .contains(&(*id, replica.clone())); + } + collections + .into_iter() + .filter_map(|(id, unhydrated)| unhydrated.then_some(id)) + .collect() + } + + /// Returns whether leader replica `replica_id` explicitly reports at least + /// one collection outside `ignored` as unhydrated. + /// + /// The leader snapshot has per-collection reports, not a separate replica + /// hydration signal. The gate uses these as a proxy: a replica is fully + /// hydrated only when all its non-ignored collections are hydrated. One + /// unhydrated collection therefore waives the incoming replica's stability + /// wait, since the leader lacks a fully hydrated counterpart. Missing + /// reports grant no exemption. + fn replica_has_unhydrated_collection( + &self, + replica_id: ReplicaId, + ignored: &BTreeSet, + ) -> bool { + let replica_id = replica_id.to_string(); + self.unhydrated_replica_collections + .iter() + .any(|(id, replica)| *replica == replica_id && !ignored.contains(id)) + } +} + #[cfg(test)] mod tests { use mz_repr::Datum; @@ -1057,6 +1175,83 @@ mod tests { use ReplicaHealth::{Exempt, HealthySince, NotHydrated, NotOnline}; + #[mz_ore::test] + fn collection_hydration_waiver_requires_complete_leader_reports() { + let frontier = |id, replica: &str| { + ( + GlobalId::User(id), + replica.to_owned(), + Antichain::from_elem(Timestamp::from(100)), + ) + }; + let row = |id: &str, replica: &str, time| { + Row::pack_slice(&[Datum::String(replica), Datum::String(id), time]) + }; + let frontiers = vec![ + frontier(1, "u10"), + frontier(1, "u11"), + frontier(2, "u10"), + frontier(2, "u11"), + frontier(3, "u10"), + frontier(3, "u11"), + frontier(4, "u10"), + frontier(5, "u10"), + ]; + let rows = vec![ + row("u1", "u10", Datum::Null), + row("u1", "u11", Datum::Null), + row("u2", "u10", Datum::Null), + row("u2", "u11", Datum::UInt64(7)), + row("u3", "u10", Datum::Null), // u11 has not reported. + row("u4", "u10", Datum::Null), + row("u4", "u12", Datum::UInt64(9)), + // u5 has no hydration reports. u6 has no frontier. + row("u6", "u10", Datum::Null), + ]; + assert_eq!( + LeaderHydration::from_rows(rows) + .collections_unhydrated_on_all_hosting_replicas(&frontiers), + BTreeSet::from([GlobalId::User(1)]) + ); + assert!( + LeaderHydration::default() + .collections_unhydrated_on_all_hosting_replicas(&frontiers) + .is_empty() + ); + } + + #[mz_ore::test] + fn replica_hydration_exemption_requires_an_unhydrated_collection_report() { + let row = |id: &str, replica: &str, time| { + Row::pack_slice(&[Datum::String(replica), Datum::String(id), time]) + }; + let leader = LeaderHydration::from_rows(vec![ + row("u1", "u10", Datum::UInt64(7)), + row("u2", "u10", Datum::UInt64(7)), + row("u1", "u11", Datum::UInt64(7)), + row("u2", "u11", Datum::Null), + row("u3", "u12", Datum::Null), + ]); + let none = BTreeSet::new(); + let replica = ReplicaId::User; + assert!( + !leader.replica_has_unhydrated_collection(replica(10), &none), + "fully hydrated" + ); + assert!(leader.replica_has_unhydrated_collection(replica(11), &none)); + assert!( + !leader.replica_has_unhydrated_collection( + replica(12), + &BTreeSet::from([GlobalId::User(3)]) + ), + "unhydrated only for an ignored collection" + ); + assert!( + !leader.replica_has_unhydrated_collection(replica(13), &none), + "no reports" + ); + } + #[mz_ore::test] fn stable_since_latest_replica() { let period_ms = 1000; diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs index 8e638873d0998..6a7beccf57d50 100644 --- a/src/catalog/src/builtin.rs +++ b/src/catalog/src/builtin.rs @@ -873,6 +873,15 @@ pub static MZ_CLUSTER_REPLICA_FRONTIERS_DESCRIPTION: LazyLock = + LazyLock::new(|| SystemObjectDescription { + schema_name: MZ_COMPUTE_HYDRATION_TIMES.schema.to_string(), + object_type: CatalogItemType::Source, + object_name: MZ_COMPUTE_HYDRATION_TIMES.name.to_string(), + }); + pub const MZ_SYSTEM_ROLE: BuiltinRole = BuiltinRole { id: MZ_SYSTEM_ROLE_ID, name: SYSTEM_USER_NAME, diff --git a/test/0dt/mock_hydration_flags.py b/test/0dt/mock_hydration_flags.py new file mode 100644 index 0000000000000..d49e927dbdf76 --- /dev/null +++ b/test/0dt/mock_hydration_flags.py @@ -0,0 +1,101 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +"""Suspend hydration on waiting's replicas and on mixed.r2.""" + +import json +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +FLAG = { + "key": "hydration-concurrency", + "version": 1, + "on": True, + "targets": [], + "prerequisites": [], + "rules": [ + { + "id": "held-replica", + "clauses": [ + { + "contextKind": "replica", + "attribute": "cluster_name", + "op": "in", + "values": ["mixed"], + "negate": False, + }, + { + "contextKind": "replica", + "attribute": "replica_name", + "op": "in", + "values": ["r2"], + "negate": False, + }, + ], + "variation": 1, + "trackEvents": False, + }, + { + "id": "held-cluster", + "clauses": [ + { + "contextKind": "replica", + "attribute": "cluster_name", + "op": "in", + "values": ["waiting"], + "negate": False, + } + ], + "variation": 1, + "trackEvents": False, + }, + ], + "fallthrough": {"variation": 0}, + "offVariation": 0, + "variations": [4, 0], + "salt": "hydration-test", + "clientSideAvailability": { + "usingMobileKey": False, + "usingEnvironmentId": False, + }, +} + + +class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_GET(self) -> None: + if self.path != "/all": + self.send_response(200 if self.path == "/health" else 404) + self.send_header("Content-Length", "0") + self.end_headers() + return + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.end_headers() + payload = {"path": "/", "data": {"flags": {FLAG["key"]: FLAG}, "segments": {}}} + try: + self.wfile.write(f"event: put\ndata: {json.dumps(payload)}\n\n".encode()) + self.wfile.flush() + while True: + time.sleep(1) + self.wfile.write(b":heartbeat\n\n") + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError): + return + + def do_POST(self) -> None: + self.rfile.read(int(self.headers.get("Content-Length", 0))) + self.send_response(202) + self.send_header("Content-Length", "0") + self.end_headers() + + +if __name__ == "__main__": + ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever() diff --git a/test/0dt/mzcompose.py b/test/0dt/mzcompose.py index c22d4133b2681..3921920c539b5 100644 --- a/test/0dt/mzcompose.py +++ b/test/0dt/mzcompose.py @@ -26,6 +26,7 @@ from materialize import buildkite from materialize.mzcompose import get_default_system_parameters, sanitizer_enabled from materialize.mzcompose.composition import Composition, Service +from materialize.mzcompose.service import Service as DockerService from materialize.mzcompose.services.kafka import Kafka from materialize.mzcompose.services.materialized import ( LEADER_STATUS_HEALTHCHECK, @@ -55,6 +56,24 @@ SYSTEM_PARAMETER_DEFAULTS = get_default_system_parameters() SERVICES = [ + DockerService( + name="hydration-flags", + config={ + "image": "python:3.12-slim", + "volumes": ["./mock_hydration_flags.py:/app/server.py"], + "command": ["python3", "-u", "/app/server.py"], + "healthcheck": { + "test": [ + "CMD", + "python3", + "-c", + "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')", + ], + "interval": "1s", + "start_period": "30s", + }, + }, + ), MySql(), Postgres(), SqlServer(), @@ -3134,6 +3153,166 @@ def workflow_caught_up_stability_survives_restart(c: Composition) -> None: c.await_mz_deployment_status(DeploymentStatus.IS_LEADER, "mz_new", sleep_time=None) +def _leader_with_hydration_flags() -> Materialized: + return Materialized( + name="mz_old", + sanity_restart=False, + deploy_generation=0, + system_parameter_defaults=SYSTEM_PARAMETER_DEFAULTS, + external_metadata_store=True, + default_replication_factor=2, + environment_extra=[ + "MZ_LAUNCHDARKLY_SDK_KEY=sdk-mock-key", + "MZ_LAUNCHDARKLY_BASE_URI=http://hydration-flags:8080", + "MZ_LAUNCHDARKLY_KEY_MAP=compute_hydration_concurrency=hydration-concurrency", + "MZ_CONFIG_SYNC_LOOP_INTERVAL=1s", + ], + ) + + +def workflow_caught_up_leader_unhydrated_collection(c: Composition) -> None: + c.down(destroy_volumes=True) + c.up("hydration-flags") + with c.override(_leader_with_hydration_flags()): + c.up("mz_old") + c.sql( + "ALTER SYSTEM SET with_0dt_caught_up_check_stability_period = '10s'", + service="mz_old", + port=6877, + user="mz_system", + ) + c.sql( + """ + CREATE CLUSTER waiting REPLICAS ( + r1 (SIZE 'scale=1,workers=1'), r2 (SIZE 'scale=1,workers=1') + ); + CREATE TABLE waiting_input (a int); + INSERT INTO waiting_input VALUES (1); + """, + service="mz_old", + ) + deadline = time.time() + 120 + while not c.sql_query( + "SELECT count(*) = 2 FROM mz_internal.mz_replica_system_parameters p " + "JOIN mz_cluster_replicas r ON r.id = p.replica_id " + "JOIN mz_clusters c ON c.id = r.cluster_id " + "WHERE c.name = 'waiting' AND p.name = 'compute_hydration_concurrency' AND p.value = '0'", + service="mz_old", + )[0][0]: + assert ( + time.time() < deadline + ), "replica hydration overrides were not applied" + time.sleep(0.5) + c.sql( + "CREATE INDEX waiting_idx IN CLUSTER waiting ON waiting_input (a)", + service="mz_old", + ) + index_id = c.sql_query( + "SELECT id FROM mz_indexes WHERE name = 'waiting_idx'", service="mz_old" + )[0][0] + deadline = time.time() + 120 + while not c.sql_query( + "SELECT count(*) = 2 AND bool_and(time_ns IS NULL) " + "FROM mz_internal.mz_compute_hydration_times " + f"WHERE object_id = '{index_id}'", + service="mz_old", + )[0][0]: + assert ( + time.time() < deadline + ), "leader did not report both replicas unhydrated" + time.sleep(0.5) + + c.up("mz_new") + c.await_mz_deployment_status( + DeploymentStatus.READY_TO_PROMOTE, "mz_new", timeout=120 + ) + with c.sql_cursor(service="mz_new", reuse_connection=False) as cursor: + cursor.execute("SET cluster = waiting") + cursor.execute("SET cluster_replica = r1") + cursor.execute("SET transaction_isolation = serializable") + cursor.execute( + ( + "SELECT hydrated_at FROM mz_introspection.mz_compute_hydration_times_per_worker " + f"WHERE export_id = '{index_id}'" + ).encode() + ) + rows = cursor.fetchall() + assert rows and all( + row[0] is None for row in rows + ), "incoming index unexpectedly hydrated" + + +def workflow_caught_up_leader_unhydrated_replica(c: Composition) -> None: + c.down(destroy_volumes=True) + c.up("hydration-flags") + with c.override(_leader_with_hydration_flags()): + c.up("mz_old") + c.sql( + "ALTER SYSTEM SET with_0dt_caught_up_check_stability_period = '10s'", + service="mz_old", + port=6877, + user="mz_system", + ) + c.sql( + """ + CREATE CLUSTER mixed REPLICAS ( + r1 (SIZE 'scale=1,workers=1'), r2 (SIZE 'scale=1,workers=1') + ); + CREATE TABLE mixed_input (a int); + INSERT INTO mixed_input VALUES (1); + """, + service="mz_old", + ) + deadline = time.time() + 120 + while not c.sql_query( + "SELECT count(*) = 1 FROM mz_internal.mz_replica_system_parameters p " + "JOIN mz_cluster_replicas r ON r.id = p.replica_id " + "JOIN mz_clusters c ON c.id = r.cluster_id " + "WHERE c.name = 'mixed' AND r.name = 'r2' " + "AND p.name = 'compute_hydration_concurrency' AND p.value = '0'", + service="mz_old", + )[0][0]: + assert time.time() < deadline, "replica hydration override was not applied" + time.sleep(0.5) + c.sql( + "CREATE INDEX mixed_idx IN CLUSTER mixed ON mixed_input (a)", + service="mz_old", + ) + index_id = c.sql_query( + "SELECT id FROM mz_indexes WHERE name = 'mixed_idx'", service="mz_old" + )[0][0] + deadline = time.time() + 120 + while c.sql_query( + "SELECT r.name, h.time_ns IS NULL FROM mz_internal.mz_compute_hydration_times h " + "JOIN mz_cluster_replicas r ON r.id = h.replica_id " + f"WHERE h.object_id = '{index_id}' ORDER BY r.name", + service="mz_old", + ) != [("r1", False), ("r2", True)]: + assert time.time() < deadline, "leader did not report mixed hydration" + time.sleep(0.5) + + c.up("mz_new") + c.await_mz_deployment_status( + DeploymentStatus.READY_TO_PROMOTE, "mz_new", timeout=120 + ) + # A hydrated r1 vetoes the collection waiver. Readiness with this + # unhydrated r2 therefore requires the separate replica exemption. + with c.sql_cursor(service="mz_new", reuse_connection=False) as cursor: + cursor.execute("SET cluster = mixed") + cursor.execute("SET cluster_replica = r2") + cursor.execute("SET transaction_isolation = serializable") + cursor.execute( + ( + "SELECT hydrated_at FROM mz_introspection.mz_compute_hydration_times_per_worker " + f"WHERE export_id = '{index_id}'" + ).encode() + ) + rows = cursor.fetchall() + assert rows and all( + row[0] is None for row in rows + ), "incoming r2 unexpectedly hydrated" + + def workflow_caught_up_stability_without_dataflows(c: Composition) -> None: """Verify the stability gate does not wait on hydration no replica reports.