Skip to content

adapter: Compare 0dt hydration against the leader #39155

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
aljoscha merged 1 commit into from
Oct 9, 2026
Merged

adapter: Compare 0dt hydration against the leader #39155

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions src/adapter/src/catalog/open/builtin_schema_migration.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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:?}");
Expand Down
225 changes: 210 additions & 15 deletions src/adapter/src/coord/caught_up.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 _;
Expand All @@ -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};
Expand Down Expand Up @@ -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,
}

Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -548,6 +569,7 @@ impl Coordinator {
cutoff.into(),
now.into(),
&live_collection_frontiers,
&leader_unhydrated_collections,
&exclude_collections,
&problematic_replicas,
)
Expand All @@ -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,
Expand All @@ -588,7 +614,12 @@ impl Coordinator {
}

/// Classifies the replicas of a cluster for the stability gate.
fn replica_targets(&self, cluster_id: ClusterId) -> BTreeMap<ReplicaId, ReplicaTarget> {
fn replica_targets(
&self,
cluster_id: ClusterId,
leader: &LeaderHydration,
ignored: &BTreeSet<GlobalId>,
) -> BTreeMap<ReplicaId, ReplicaTarget> {
let cluster = self.catalog().get_cluster(cluster_id);
let has_hydration_log = cluster
.log_indexes
Expand All @@ -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
Expand All @@ -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`.
Expand All @@ -643,6 +678,7 @@ impl Coordinator {
cutoff: Timestamp,
now: Timestamp,
live_frontiers: &BTreeMap<GlobalId, Antichain<Timestamp>>,
leader_unhydrated_collections: &BTreeSet<GlobalId>,
exclude_collections: &BTreeSet<GlobalId>,
problematic_replicas: &BTreeSet<ReplicaId>,
) -> BTreeMap<ClusterId, ClusterCaughtUpStatus> {
Expand All @@ -655,6 +691,7 @@ impl Coordinator {
cutoff.clone(),
now.clone(),
live_frontiers,
leader_unhydrated_collections,
exclude_collections,
problematic_replicas,
)
Expand Down Expand Up @@ -688,6 +725,7 @@ impl Coordinator {
cutoff: Timestamp,
now: Timestamp,
live_frontiers: &BTreeMap<GlobalId, Antichain<Timestamp>>,
leader_unhydrated_collections: &BTreeSet<GlobalId>,
exclude_collections: &BTreeSet<GlobalId>,
problematic_replicas: &BTreeSet<ReplicaId>,
) -> Result<ClusterCaughtUpStatus, anyhow::Error> {
Expand Down Expand Up @@ -851,16 +889,19 @@ 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)),
);

// 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);
Expand Down Expand Up @@ -1048,6 +1089,83 @@ impl Coordinator {
}
}

/// The leader's hydration reports, read from its `mz_compute_hydration_times`.

@aljoscha aljoscha Oct 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this across all replicas of some cluster, so the unhydrated/hydrated is mixed, or hydration status of one specific replica. Need to be specific in the code docs and naming here, please

#[derive(Debug, Default)]
struct LeaderHydration {
hydrated_on_any_replica: BTreeSet<GlobalId>,
/// (Collection ID, replica ID) pairs explicitly reported as unhydrated.
unhydrated_replica_collections: BTreeSet<(GlobalId, String)>,
}

impl LeaderHydration {
fn from_rows(rows: Vec<Row>) -> 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<Timestamp>)],
) -> BTreeSet<GlobalId> {
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<GlobalId>,
) -> 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;
Expand All @@ -1057,6 +1175,83 @@ mod tests {

use ReplicaHealth::{Exempt, HealthySince, NotHydrated, NotOnline};

#[mz_ore::test]
fn collection_hydration_waiver_requires_complete_leader_reports() {

@ggevay ggevay Oct 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nothing in the 0dt suite creates a collection or replica the leader hasn't hydrated, and CI runs with a 0 s stability period, so a no-op would pass. Could we add a workflow for each rule (e.g. a leader whose hydration of one object is stalled, and the crash-loop workflow with a replica the leader hasn't fully hydrated)?

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;
Expand Down
9 changes: 9 additions & 0 deletions src/catalog/src/builtin.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -873,6 +873,15 @@ pub static MZ_CLUSTER_REPLICA_FRONTIERS_DESCRIPTION: LazyLock<SystemObjectDescri
object_type: CatalogItemType::Source,
object_name: MZ_CLUSTER_REPLICA_FRONTIERS.name.to_string(),
});

/// Identifies the leader hydration source that the 0dt caught-up gate reads.
pub static MZ_COMPUTE_HYDRATION_TIMES_DESCRIPTION: LazyLock<SystemObjectDescription> =
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,
Expand Down
Loading
Loading