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.