catalog: sync builtin cluster replica sizes and add in-use check

During catalog open, sync the env-var cluster replica sizes into
the durable catalog as builtin entries. This ensures sizes survive
restarts and can be managed alongside user-defined sizes.

Key changes:
- `sync_builtin_cluster_replica_sizes()` reconciles env-var sizes
  with durable state: inserts new sizes, updates changed allocations,
  and removes sizes no longer in the env var.
- `CatalogState.cluster_replica_sizes` is now initialized empty and
  populated from durable state updates (via apply_cluster_replica_size_update
  from Commit 1), rather than directly from config.
- `is_cluster_replica_size_in_use()` helper scans all cluster replicas
  to check if a given size name is referenced. This will be used in
  the DDL drop path (Commit 3) to prevent deletion of in-use sizes.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jubrad

src/adapter/src/catalog/open.rs

@@ -29,7 +29,7 @@ use mz_catalog::builtin::{
     BUILTIN_CLUSTER_REPLICAS, BUILTIN_CLUSTERS, BUILTIN_PREFIXES, BUILTIN_ROLES, BUILTINS, Builtin,
     Fingerprint, MZ_CATALOG_RAW, RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL,
 };
-use mz_catalog::config::StateConfig;
+use mz_catalog::config::{ClusterReplicaSizeMap, StateConfig};
 use mz_catalog::durable::objects::{
     SystemObjectDescription, SystemObjectMapping, SystemObjectUniqueIdentifier,
 };
@@ -176,7 +176,8 @@ impl Catalog {
                 },
                 helm_chart_version: config.helm_chart_version,
             },
-            cluster_replica_sizes: config.cluster_replica_sizes,
+            // Start empty; populated from durable state updates after builtin sync.
+            cluster_replica_sizes: ClusterReplicaSizeMap(std::collections::BTreeMap::new()),
             availability_zones: config.availability_zones,
             egress_addresses: config.egress_addresses,
             aws_principal_context: config.aws_principal_context,
@@ -228,6 +229,7 @@ impl Catalog {
                 config.boot_ts,
             )?;
             add_new_remove_old_builtin_roles_migration(&mut txn)?;
+            sync_builtin_cluster_replica_sizes(&mut txn, &config.cluster_replica_sizes)?;
             remove_invalid_config_param_role_defaults_migration(&mut txn)?;
             remove_pending_cluster_replicas_migration(&mut txn, config.boot_ts)?;
 
@@ -1118,6 +1120,57 @@ fn add_new_remove_old_builtin_roles_migration(
     Ok(())
 }
 
+/// Syncs the builtin cluster replica sizes from the env-var configuration into
+/// the durable catalog. Inserts new sizes, updates changed sizes, and removes
+/// sizes that are no longer in the configuration.
+fn sync_builtin_cluster_replica_sizes(
+    txn: &mut Transaction<'_>,
+    config_sizes: &ClusterReplicaSizeMap,
+) -> Result<(), AdapterError> {
+    use std::collections::BTreeMap;
+
+    use mz_catalog::durable::objects::DurableType;
+
+    // Collect current durable builtin sizes.
+    let durable_sizes: BTreeMap<String, _> = txn
+        .get_cluster_replica_sizes()
+        .filter(|s| s.builtin)
+        .map(|s| (s.name.clone(), s))
+        .collect();
+
+    // Insert or update builtins from config.
+    for (name, allocation) in &config_sizes.0 {
+        if let Some(existing) = durable_sizes.get(name) {
+            // Check if the allocation changed by comparing the durable value
+            // representation (which is Eq-comparable).
+            let new_size = mz_catalog::durable::ClusterReplicaSize {
+                name: name.clone(),
+                allocation: allocation.clone(),
+                builtin: true,
+            };
+            let (_, new_value) = DurableType::into_key_value(new_size);
+            let (_, existing_value) = DurableType::into_key_value(existing.clone());
+            if new_value != existing_value {
+                // Retract and reinsert with updated allocation.
+                txn.remove_cluster_replica_size(name)?;
+                txn.insert_cluster_replica_size(name.clone(), allocation.clone(), true)?;
+            }
+        } else {
+            // New builtin size, insert it.
+            txn.insert_cluster_replica_size(name.clone(), allocation.clone(), true)?;
+        }
+    }
+
+    // Remove builtins that are no longer in config.
+    for (name, _) in &durable_sizes {
+        if !config_sizes.0.contains_key(name) {
+            txn.remove_cluster_replica_size(name)?;
+        }
+    }
+
+    Ok(())
+}
+
 fn add_new_remove_old_builtin_cluster_replicas_migration(
     txn: &mut Transaction<'_>,
     builtin_cluster_config_map: &BuiltinBootstrapClusterConfigMap,

src/adapter/src/catalog/state.rs

@@ -2446,6 +2446,20 @@ impl CatalogState {
         }
     }
 
+    /// Returns true if the given cluster replica size name is in use by any
+    /// existing cluster replica.
+    pub fn is_cluster_replica_size_in_use(&self, size_name: &str) -> bool {
+        use mz_controller::clusters::ReplicaLocation;
+        self.clusters_by_id.values().any(|cluster| {
+            cluster.replicas().any(|replica| {
+                matches!(
+                    &replica.config.location,
+                    ReplicaLocation::Managed(loc) if loc.size == size_name
+                )
+            })
+        })
+    }
+
     pub fn ensure_not_reserved_role(&self, role_id: &RoleId) -> Result<(), Error> {
         if role_id.is_builtin() {
             let role = self.get_role(role_id);