adapter: hydrate migrated builtin MVs before 0dt cut-over

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.

The behaviour is gated behind the `enable_0dt_hydrate_migrated_builtin_mvs`
feature flag, on by default, a break-glass revert to the prior
exclude-from-gate behaviour without a redeploy.

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,10 @@ 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.
+
 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

@@ -744,6 +744,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

@@ -3289,6 +3289,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",

misc/scratch/ember-scratch.json

@@ -0,0 +1,8 @@
+{
+  "name": "ember-scratch",
+  "launch_script": "true",
+  "instance_type": "c7a.12xlarge",
+  "ami": "ami-0ecb62995f68bb549",
+  "size_gb": 512,
+  "tags": {}
+}

src/adapter-types/src/dyncfgs.rs

@@ -80,6 +80,16 @@ pub const WITH_0DT_CAUGHT_UP_CHECK_STABILITY_PERIOD: Config<Duration> = Config::
     "How long a cluster must continuously be caught-up and have all replicas healthy before it is considered ready to cut over during a 0dt deployment.",
 );
 
+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 reverts to excluding migrated MVs (and their dependents) from the caught-up \
+     check, which lets promotion proceed with them unhydrated and hydrate at cut-over instead. \
+     Only takes effect when the leader is new enough for the write to make progress.",
+);
+
 /// 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",
@@ -458,6 +468,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/coord.rs

@@ -93,7 +93,8 @@ use mz_adapter_types::compaction::CompactionWindow;
 use mz_adapter_types::connection::ConnectionId;
 use mz_adapter_types::dyncfgs::FRONTEND_READ_THEN_WRITE;
 use mz_adapter_types::dyncfgs::{
-    USER_ID_POOL_BATCH_SIZE, WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL,
+    ENABLE_0DT_HYDRATE_MIGRATED_BUILTIN_MVS, USER_ID_POOL_BATCH_SIZE,
+    WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL,
 };
 use mz_auth::password::Password;
 use mz_build_info::BuildInfo;
@@ -171,6 +172,7 @@ use mz_storage_types::sources::{IngestionDescription, SourceExport, Timeline};
 use mz_timestamp_oracle::{TimestampOracleConfig, WriteTimestamp};
 use mz_transform::dataflow::DataflowMetainfo;
 use opentelemetry::trace::TraceContextExt;
+use semver::Version;
 use serde::Serialize;
 use thiserror::Error;
 use timely::progress::{Antichain, Timestamp as _};
@@ -243,6 +245,17 @@ mod privatelink_status;
 mod sql;
 mod validity;
 
+/// The oldest leader version against which a replacement-migrated builtin materialized view may
+/// write its new persist shard while this environment is still read-only.
+///
+/// Every builtin materialized view reads `mz_internal.mz_catalog_raw`, so its dataflow only makes
+/// progress up to the catalog shard's frontier. Holding that frontier at the current time is the
+/// leader's job, and leaders only started doing it in v26.17. Write-enable such an MV against an
+/// older leader and it sits at a stale frontier and never reports caught up, which blocks
+/// promotion outright instead of merely leaving the collection cold at cut-over. We still support
+/// upgrading from before v26.17, so that leader is a real case, not a hypothetical.
+const MIN_LEADER_VERSION_FOR_MIGRATED_MV_WRITES: Version = Version::new(26, 17, 0);
+
 /// A pool of pre-allocated user IDs to avoid per-DDL persist writes.
 ///
 /// IDs in the range `[next, upper)` are available for allocation.
@@ -2445,6 +2458,7 @@ impl Coordinator {
         &mut self,
         boot_ts: Timestamp,
         migrated_storage_collections_0dt: BTreeSet<CatalogItemId>,
+        hydrate_migrated_mvs: bool,
         mut builtin_table_updates: Vec<BuiltinTableUpdate>,
         cached_global_exprs: BTreeMap<GlobalId, GlobalExpressions>,
         uncached_local_exprs: BTreeMap<GlobalId, LocalExpressions>,
@@ -2773,7 +2787,22 @@ impl Coordinator {
                     // If this is a replacement MV, it must remain read-only until the replacement
                     // gets applied.
                     if mview.replacement_target.is_none() {
-                        self.allow_writes(mview.cluster_id, mview.global_id_writes());
+                        let gid = mview.global_id_writes();
+                        if hydrate_migrated_mvs
+                            && migrated_storage_collections_0dt.contains(&entry.id())
+                        {
+                            // `migrated_storage_collections_0dt` is `Replacement`-migrated items
+                            // only, so this is a fresh shard we own: nothing else writes it, and
+                            // writing it while read-only hydrates the MV and its dependents before
+                            // cut-over. An `Evolution`-migrated MV reuses the leader's live shard
+                            // and must never reach here.
+                            self.controller
+                                .compute
+                                .allow_writes_in_read_only(mview.cluster_id, gid)
+                                .unwrap_or_terminate("allow_writes cannot fail");
+                        } else {
+                            self.allow_writes(mview.cluster_id, gid);
+                        }
                     }
                 }
                 CatalogItem::MetricSink(metric_sink) => {
@@ -5021,6 +5050,7 @@ pub fn serve(
         ;
         let OpenCatalogResult {
             mut catalog,
+            last_seen_version,
             migrated_storage_collections_0dt,
             new_builtin_collections,
             builtin_table_updates,
@@ -5076,6 +5106,22 @@ pub fn serve(
             catalog_open_start.elapsed()
         );
 
+        // Whether replacement-migrated builtin MVs may write their new shards before cut-over.
+        // Both `bootstrap` and the readiness gate below read this, and they have to agree.
+        // `MIN_LEADER_VERSION_FOR_MIGRATED_MV_WRITES` explains why the leader's version settles it.
+        //
+        // While we are read-only, `last_seen_version` is that leader's version: our catalog
+        // transaction is a savepoint, so our own bump of the setting never lands. `None` means a
+        // freshly initialized catalog, with nothing migrated and no leader to be compatible with.
+        //
+        // `ENABLE_0DT_HYDRATE_MIGRATED_BUILTIN_MVS` is the break-glass revert: off falls back to
+        // excluding migrated MVs from the caught-up gate, no redeploy needed.
+        let hydrate_migrated_mvs = ENABLE_0DT_HYDRATE_MIGRATED_BUILTIN_MVS
+            .get(catalog.system_config().dyncfgs())
+            && last_seen_version
+                .as_ref()
+                .is_none_or(|version| *version >= MIN_LEADER_VERSION_FOR_MIGRATED_MV_WRITES);
+
         let coord_thread_start = Instant::now();
         info!("startup: coordinator init: coordinator thread start beginning");
 
@@ -5126,18 +5172,16 @@ pub fn serve(
 
                 // A collection that can't advance its write frontier in read-only mode
                 // stalls its transitive dependents too, so exclude those from the caught-up
-                // check as well. That's migrated MVs (their dataflows don't write in
-                // read-only mode) and new builtin MVs (their fresh shard has no writer until
-                // this deployment promotes). An excluded dependent may still be hydrating
-                // right after promotion, a brief blip we accept because these MVs are small
-                // and get a writer at cut-over.
+                // check as well. That's new builtin MVs, whose fresh shard has no writer until
+                // this deployment promotes, plus migrated MVs whenever the leader is too old for
+                // them to write. An excluded dependent may still be hydrating right after
+                // promotion, a brief blip we accept because these MVs are small and get a writer
+                // at cut-over.
                 //
-                // TODO: Consider sending `allow_writes` for the dataflows of migrated MVs, which
-                //       would allow them to make progress even in read-only mode. This doesn't
-                //       work for MVs based on `mz_catalog_raw`, if the leader's version is less
-                //       than v26.17, since before that version the catalog shard's frontier wasn't
-                //       kept up-to-date with the current time. So this workaround has to remain in
-                //       place upgrades from a version less than v26.17 are no longer supported.
+                // A migrated builtin *table* needs no such treatment even though a builtin MV can
+                // read one (`mz_clusters` joins `mz_cluster_replica_size_internal`):
+                // `read_only_mode_table_worker` keeps advancing the uppers of migrated tables, so
+                // an MV over one still catches up.
                 let new_builtin_mvs = new_builtin_collections
                     .iter()
                     .map(|global_id| {
@@ -5148,12 +5192,12 @@ pub fn serve(
                     })
                     .filter(|entry| entry.is_materialized_view())
                     .map(|entry| entry.id());
-                let mut todo: Vec<_> = migrated_storage_collections_0dt
+                let frozen_migrated_mvs = migrated_storage_collections_0dt
                     .iter()
                     .copied()
-                    .filter(|id| catalog.state().get_entry(id).is_materialized_view())
-                    .chain(new_builtin_mvs)
-                    .collect();
+                    .filter(|_| !hydrate_migrated_mvs)
+                    .filter(|id| catalog.state().get_entry(id).is_materialized_view());
+                let mut todo: Vec<_> = new_builtin_mvs.chain(frozen_migrated_mvs).collect();
                 while let Some(item_id) = todo.pop() {
                     let entry = catalog.state().get_entry(&item_id);
                     exclude_collections.extend(entry.global_ids());
@@ -5326,6 +5370,7 @@ pub fn serve(
                         .bootstrap(
                             boot_ts,
                             migrated_storage_collections_0dt,
+                            hydrate_migrated_mvs,
                             builtin_table_updates,
                             cached_global_exprs,
                             uncached_local_exprs,