adapter: hydrate migrated builtin MVs before 0dt cut-over (#38399)

Problem:
A builtin schema migration using the `Replacement` mechanism hands the
new deployment a fresh persist shard. Nothing writes that shard while
the deployment is read-only, because `ComputeController::allow_writes`
no-ops in read-only mode, so the MV's write frontier never advances and
the 0dt readiness gate has to drop it and everything downstream of it
from the caught-up check. The deployment then promotes with those
collections unhydrated, and they all hydrate at once at cut-over,
spiking catalog-server CPU and degrading catalog queries.

Solution:
The replacement shard is exclusively owned by this deployment, so the MV
can write it while we are still read-only. Do that, and keep the MV in
the readiness gate, so it hydrates before cut-over instead of at it.

Previously we didn't do this for builtins derived from mz_catalog
because catalog shard frontier was not held. However, as of 26.17 it is,
so gate the behaviour on the version of the old leader.

Testing:
- New environmentd integration test
`test_0dt_migrated_builtin_mv_hydrates_before_promotion` boots a
read-only generation with a forced `replacement` migration and asserts
the migrated builtin MVs `mz_databases` and `mz_clusters` are readable
before it reports `ReadyToPromote`, proving they hydrate before cut-over
rather than at it.
- Extended the `0dt` mzcompose workflow
`builtin-schema-migrations-replacement` to read those MVs from the
read-only generation before promotion, guarding the same invariant
through a real cross-version upgrade.

Author: mtabebe

doc/developer/design/20251015_builtin_schema_migration.md

@@ -58,7 +58,8 @@ Doing so requires no writes to durable state, and therefore doesn't interfere wi
 
 In the subsequent read-only bootstrap phase, the process creates persist read and write handles using the new schema.
 Read handles perform transparent migration of any data updates that flow through them, so dataflow hydration can proceed using the new schema.
-Write handles only require a matching registered shard schema when writing batches, which is something a read-only process doesn't do.
+Write handles only require a matching registered shard schema when writing batches, which a read-only process performing schema evolution doesn't do: the shard is the leader's live one, so it stays read-only until promotion.
+(Shard replacement is different: there the read-only process does write batches, but to a shard it created for itself. See below.)
 
 Once the read-only process gets promoted to a leader, and runs the builtin schema migration mechanism again, it this time registers the new schema with the persist shard.
 Doing so fences out any processes that planned to evolve the schema to an earlier version.
@@ -76,6 +77,15 @@ Notably, it ignores any entries at different versions or deploy generations, to
 Depending on the existing migration shard entries, the process either decides to use the existing replacement shard, or to create the replacement shard and write its ID into the migration shard, at the current version.
 It sets the new shard ID as the migrated collection's shard in its in-memory catalog and commences bootstrapping using the replacement shard.
 
+Because this environment exclusively owns the replacement shard, the read-only process force-writes it during bootstrap rather than leaving it read-only until promotion.
+This lets a migrated builtin materialized view and its dependents hydrate before cut-over instead of all at once at cut-over.
+It is safe only for the self-owned replacement shard, never a shard the leader is still serving, which is why it applies to shard replacement and not schema evolution.
+
+The force-write is conditional on two things.
+First, the leader must be at v26.17 or later, because every builtin materialized view reads the catalog shard and only leaders from that version on keep its frontier advancing with the current time; against an older leader the dataflow would sit at a stale frontier.
+Second, the `enable_0dt_hydrate_migrated_builtin_mvs` feature flag must be on; it exists as a break-glass revert.
+When either condition does not hold, the migrated materialized views and their dependents are instead excluded from the 0dt caught-up check, which is the older behaviour: promotion proceeds without them and they hydrate at cut-over.
+
 A leader process performing shard replacement performs the same steps as in read-only mode.
 Additionally, it cleans up durable state written by earlier versions and/or deploy generations by:
   - arranging for the previous shards used by the migrated storage collections to be finalized

misc/python/materialize/mzcompose/__init__.py

@@ -750,6 +750,7 @@ def get_default_system_parameters(
     "with_0dt_caught_up_check_cutoff",
     "enable_0dt_caught_up_replica_status_check",
     "enable_0dt_caught_up_stability_check",
+    "enable_0dt_hydrate_migrated_builtin_mvs",
     "plan_insights_notice_fast_path_clusters_optimize_duration",
     "enable_expression_cache",
     "mz_metrics_lgalloc_map_refresh_interval",

misc/python/materialize/parallel_workload/action.py

@@ -3284,6 +3284,7 @@ def __init__(
             "with_0dt_caught_up_check_cutoff",
             "with_0dt_caught_up_check_stability_period",
             "enable_0dt_caught_up_stability_check",
+            "enable_0dt_hydrate_migrated_builtin_mvs",
             "enable_statement_lifecycle_logging",
             "enable_introspection_subscribes",
             "plan_insights_notice_fast_path_clusters_optimize_duration",

src/adapter-types/src/dyncfgs.rs

@@ -90,6 +90,19 @@ pub const WITH_0DT_CAUGHT_UP_CHECK_STABILITY_PERIOD: Config<Duration> = Config::
     ParameterScope::Environment,
 );
 
+pub const ENABLE_0DT_HYDRATE_MIGRATED_BUILTIN_MVS: Config<bool> = Config::new(
+    "enable_0dt_hydrate_migrated_builtin_mvs",
+    true,
+    "Write-enable replacement-migrated builtin materialized views while read-only during a 0dt \
+     deployment, so they hydrate before cut-over and keep gating promotion. Emergency break-glass \
+     flag: disabling excludes migrated MVs (and their dependents) from the caught-up check again, \
+     so promotion proceeds with them unhydrated. Not an exact revert: a collection with no live \
+     leader frontier must be hydrated either way. Only takes effect when the leader is new enough \
+     for the write to make progress, and is read once at startup, so changing it means setting it \
+     on the leader and restarting the new deployment.",
+    ParameterScope::Environment,
+);
+
 /// Enable logging of statement lifecycle events in mz_internal.mz_statement_lifecycle_history.
 pub const ENABLE_STATEMENT_LIFECYCLE_LOGGING: Config<bool> = Config::new(
     "enable_statement_lifecycle_logging",
@@ -511,6 +524,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&ENABLE_0DT_CAUGHT_UP_REPLICA_STATUS_CHECK)
         .add(&ENABLE_0DT_CAUGHT_UP_STABILITY_CHECK)
         .add(&WITH_0DT_CAUGHT_UP_CHECK_STABILITY_PERIOD)
+        .add(&ENABLE_0DT_HYDRATE_MIGRATED_BUILTIN_MVS)
         .add(&ENABLE_STATEMENT_LIFECYCLE_LOGGING)
         .add(&ENABLE_INTROSPECTION_SUBSCRIBES)
         .add(&ENABLE_FRONTEND_SUBSCRIBES)

src/adapter/src/catalog.rs

@@ -648,6 +648,7 @@ impl Catalog {
 
         let OpenCatalogResult {
             catalog,
+            last_seen_version: _,
             migrated_storage_collections_0dt: _,
             new_builtin_collections: _,
             builtin_table_updates: _,

src/adapter/src/catalog/open.rs

@@ -61,6 +61,7 @@ use mz_sql::session::user::{MZ_SYSTEM_ROLE_ID, SYSTEM_USER};
 use mz_sql::session::vars::{SessionVars, SystemVars, VarError, VarInput};
 use mz_storage_client::controller::{StorageMetadata, StorageTxn};
 use mz_storage_client::storage_collections::StorageCollections;
+use semver::Version;
 use tracing::{Instrument, info, warn};
 use uuid::Uuid;
 
@@ -73,14 +74,26 @@ use crate::catalog::{BuiltinTableUpdate, Catalog, CatalogState, Config, is_reser
 pub struct InitializeStateResult {
     /// An initialized [`CatalogState`].
     pub state: CatalogState,
-    /// A set of new shards that may need to be initialized (only used by 0dt migration).
+    /// Items whose builtin schema migration allocated a fresh, self-owned persist shard. Only used
+    /// by 0dt migration.
+    ///
+    /// `Replacement`-migrated items only: derived by filtering to items with a global id in
+    /// `MigrationRunResult::new_shards`, which only `migrate_replace` populates. `Evolution`
+    /// migrates in place and reuses the leader's shard, so it never lands here. Read-only write
+    /// paths depend on that: force-writing a shard while read-only is safe only because we
+    /// exclusively own it (see `ComputeController::allow_writes_in_read_only`).
     pub migrated_storage_collections_0dt: BTreeSet<CatalogItemId>,
     /// A set of new builtin items.
     pub new_builtin_collections: BTreeSet<GlobalId>,
     /// A list of builtin table updates corresponding to the initialized state.
     pub builtin_table_updates: Vec<BuiltinTableUpdate>,
-    /// The version of the catalog that existed before initializing the catalog.
-    pub last_seen_version: String,
+    /// The version of the binary that last committed catalog migrations, or `None` for a newly
+    /// initialized catalog.
+    ///
+    /// While this environment is read-only during a 0dt deployment, this is the version of the
+    /// leader environment: the read-only catalog transaction is a savepoint, so our own bump of
+    /// the setting never lands.
+    pub last_seen_version: Option<Version>,
     /// A handle to the expression cache if it's enabled.
     pub expr_cache_handle: Option<ExpressionCacheHandle>,
     /// The global expressions that were cached in `expr_cache_handle`.
@@ -92,7 +105,10 @@ pub struct InitializeStateResult {
 pub struct OpenCatalogResult {
     /// An opened [`Catalog`].
     pub catalog: Catalog,
-    /// A set of new shards that may need to be initialized.
+    /// See [`InitializeStateResult::last_seen_version`].
+    pub last_seen_version: Option<Version>,
+    /// See [`InitializeStateResult::migrated_storage_collections_0dt`]; `Replacement`-migrated
+    /// items only.
     pub migrated_storage_collections_0dt: BTreeSet<CatalogItemId>,
     /// A set of new builtin items.
     pub new_builtin_collections: BTreeSet<GlobalId>,
@@ -430,8 +446,7 @@ impl Catalog {
             .await;
         builtin_table_updates.extend(builtin_table_update);
 
-        let last_seen_version =
-            get_migration_version(&txn).map_or_else(|| "new".into(), |v| v.to_string());
+        let last_seen_version = get_migration_version(&txn);
 
         let mz_authentication_mock_nonce =
             txn.get_authentication_mock_nonce().ok_or_else(|| {
@@ -453,7 +468,9 @@ impl Catalog {
             .await
             .map_err(|e| {
                 Error::new(ErrorKind::FailedCatalogMigration {
-                    last_seen_version: last_seen_version.clone(),
+                    last_seen_version: last_seen_version
+                        .as_ref()
+                        .map_or_else(|| "new".to_string(), |v| v.to_string()),
                     this_version: config.build_info.version,
                     cause: e.to_string(),
                 })
@@ -566,7 +583,7 @@ impl Catalog {
                 migrated_storage_collections_0dt,
                 new_builtin_collections,
                 mut builtin_table_updates,
-                last_seen_version: _,
+                last_seen_version,
                 expr_cache_handle,
                 cached_global_exprs,
                 uncached_local_exprs,
@@ -624,6 +641,7 @@ impl Catalog {
 
             Ok(OpenCatalogResult {
                 catalog,
+                last_seen_version,
                 migrated_storage_collections_0dt,
                 new_builtin_collections,
                 builtin_table_updates,

src/adapter/src/catalog/open/builtin_schema_migration.rs

@@ -37,8 +37,9 @@ 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_OBJECT_ARRANGEMENT_SIZE_HISTORY_DESCRIPTION, MZ_STORAGE_USAGE_BY_SHARD,
-    MZ_STORAGE_USAGE_BY_SHARD_DESCRIPTION, RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL,
+    MZ_CLUSTER_REPLICA_FRONTIERS_DESCRIPTION, MZ_OBJECT_ARRANGEMENT_SIZE_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;
@@ -758,6 +759,15 @@ impl Migration {
                 "mz_catalog_raw cannot be migrated"
             );
 
+            // The 0dt caught-up gate reads the leader's `mz_cluster_replica_frontiers` shard for
+            // the live frontiers it checks every collection against. Migrating it via `Replacement`
+            // hands us a fresh shard we write ourselves, so the gate would compare us against
+            // ourselves instead of against the leader.
+            assert_ne!(
+                &*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"
+            );
+
             let Some(object_info) = self.system_objects.get(object) else {
                 panic!("migration step for non-existent builtin: {object:?}");
             };