catalog: add durable storage for cluster replica sizes

Add ClusterReplicaSize as a new durable catalog object type. This
lays the groundwork for making cluster replica sizes persistent and
user-definable via SQL DDL.

The new type follows the established pattern (NetworkPolicy, Cluster,
etc.):

- Proto Key/Value definitions in catalog-protos with Arbitrary derives
- DurableType impl converting between ReplicaAllocation and raw
  proto-compatible fields (the Value stores raw u64/string fields
  rather than ReplicaAllocation directly, since Numeric doesn't
  implement Eq/Ord needed by TableTransaction)
- StateUpdateKind variant with collection type mapping
- Transaction CRUD methods (insert, remove, get)
- Snapshot field and persist read/write support
- Memory StateUpdateKind variant applied in pre-cluster ordering
- Debug trace support
- No-op v81→v82 catalog migration (existing catalogs have no
  ClusterReplicaSize entries; builtins will be populated in a
  follow-up commit)

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

Author: jubrad

src/adapter/src/catalog/apply.rs

@@ -300,6 +300,9 @@ impl CatalogState {
             StateUpdateKind::NetworkPolicy(network_policy) => {
                 self.apply_network_policy_update(network_policy, diff, retractions);
             }
+            StateUpdateKind::ClusterReplicaSize(size) => {
+                self.apply_cluster_replica_size_update(size, diff, retractions);
+            }
             StateUpdateKind::IntrospectionSourceIndex(introspection_source_index) => {
                 self.apply_introspection_source_index_update(
                     introspection_source_index,
@@ -542,6 +545,25 @@ impl CatalogState {
         );
     }
 
+    #[instrument(level = "debug")]
+    fn apply_cluster_replica_size_update(
+        &mut self,
+        size: mz_catalog::durable::ClusterReplicaSize,
+        diff: StateDiff,
+        _retractions: &mut InProgressRetractions,
+    ) {
+        match diff {
+            StateDiff::Addition => {
+                self.cluster_replica_sizes
+                    .0
+                    .insert(size.name, size.allocation);
+            }
+            StateDiff::Retraction => {
+                self.cluster_replica_sizes.0.remove(&size.name);
+            }
+        }
+    }
+
     #[instrument(level = "debug")]
     fn apply_introspection_source_index_update(
         &mut self,
@@ -1457,6 +1479,7 @@ impl CatalogState {
             StateUpdateKind::Database(_)
             | StateUpdateKind::Schema(_)
             | StateUpdateKind::NetworkPolicy(_)
+            | StateUpdateKind::ClusterReplicaSize(_)
             | StateUpdateKind::StorageCollectionMetadata(_)
             | StateUpdateKind::UnfinalizedShard(_) => Vec::new(),
         }
@@ -2031,7 +2054,8 @@ fn sort_updates(updates: Vec<StateUpdate>) -> Vec<StateUpdate> {
             | StateUpdateKind::DefaultPrivilege(_)
             | StateUpdateKind::SystemPrivilege(_)
             | StateUpdateKind::SystemConfiguration(_)
-            | StateUpdateKind::NetworkPolicy(_) => push_update(
+            | StateUpdateKind::NetworkPolicy(_)
+            | StateUpdateKind::ClusterReplicaSize(_) => push_update(
                 update,
                 diff,
                 &mut pre_cluster_retractions,
@@ -2397,6 +2421,7 @@ impl ApplyState {
             | SystemConfiguration(_)
             | Cluster(_)
             | NetworkPolicy(_)
+            | ClusterReplicaSize(_)
             | ClusterReplica(_)
             | SourceReferences(_)
             | Comment(_)

src/adapter/src/catalog/open.rs

@@ -282,6 +282,7 @@ impl Catalog {
                 | BootstrapStateUpdateKind::SystemConfiguration(_)
                 | BootstrapStateUpdateKind::Cluster(_)
                 | BootstrapStateUpdateKind::NetworkPolicy(_)
+                | BootstrapStateUpdateKind::ClusterReplicaSize(_)
                 | BootstrapStateUpdateKind::ClusterReplica(_) => {
                     pre_item_updates.push(StateUpdate {
                         kind: kind.into(),

src/catalog-protos/objects_hashes.json

@@ -1,7 +1,7 @@
 [
   {
     "name": "objects.rs",
-    "md5": "42585baa1b4b5e1b6da4e361a35ec546"
+    "md5": "11000173b41468f6ff3f9593b31e5db7"
   },
   {
     "name": "objects_v74.rs",
@@ -34,5 +34,9 @@
   {
     "name": "objects_v81.rs",
     "md5": "42585baa1b4b5e1b6da4e361a35ec546"
+  },
+  {
+    "name": "objects_v82.rs",
+    "md5": "11000173b41468f6ff3f9593b31e5db7"
   }
 ]

src/catalog-protos/src/lib.rs

@@ -19,14 +19,15 @@ pub mod objects_v78;
 pub mod objects_v79;
 pub mod objects_v80;
 pub mod objects_v81;
+pub mod objects_v82;
 pub mod serialization;
 
 /// The current version of the `Catalog`.
 ///
 /// We will initialize new `Catalog`s with this version, and migrate existing `Catalog`s to this
 /// version. Whenever the `Catalog` changes, e.g. the types we serialize in the `Catalog`
 /// change, we need to bump this version.
-pub const CATALOG_VERSION: u64 = 81;
+pub const CATALOG_VERSION: u64 = 82;
 
 /// The minimum `Catalog` version number that we support migrating from.
 ///

src/catalog-protos/src/objects.rs

@@ -465,6 +465,63 @@ pub struct NetworkPolicyValue {
     pub oid: u32,
 }
 
+#[derive(
+    Clone,
+    Debug,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Serialize,
+    Deserialize,
+    Arbitrary
+)]
+pub struct ClusterReplicaSizeKey {
+    pub name: String,
+}
+
+#[derive(
+    Clone,
+    Debug,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Serialize,
+    Deserialize,
+    Arbitrary
+)]
+pub struct ClusterReplicaSizeValue {
+    /// The memory limit for each process, in bytes.
+    pub memory_limit: Option<u64>,
+    /// The memory request for each process, in bytes.
+    pub memory_request: Option<u64>,
+    /// The CPU limit for each process, in nanocpus.
+    pub cpu_limit: Option<u64>,
+    /// The CPU request for each process, in nanocpus.
+    pub cpu_request: Option<u64>,
+    /// The disk limit for each process, in bytes.
+    pub disk_limit: Option<u64>,
+    /// The number of processes in the replica.
+    pub scale: u16,
+    /// The number of worker threads per process.
+    pub workers: u64,
+    /// The number of credits per hour that the replica consumes.
+    pub credits_per_hour: String,
+    /// Whether each process has exclusive access to its CPU cores.
+    pub cpu_exclusive: bool,
+    /// Whether this size represents a modern "cc" size rather than a legacy T-shirt size.
+    pub is_cc: bool,
+    /// Whether instances of this type use swap as the spill-to-disk mechanism.
+    pub swap_enabled: bool,
+    /// Whether instances of this type can be created.
+    pub disabled: bool,
+    /// Additional node selectors.
+    pub selectors: std::collections::BTreeMap<String, String>,
+    /// Whether this is a builtin size (from env var) or user-defined.
+    pub builtin: bool,
+}
+
 #[derive(
     Clone,
     Debug,
@@ -2514,6 +2571,7 @@ pub enum StateUpdateKind {
     Cluster(Cluster),
     ClusterIntrospectionSourceIndex(ClusterIntrospectionSourceIndex),
     ClusterReplica(ClusterReplica),
+    ClusterReplicaSize(ClusterReplicaSize),
     Comment(Comment),
     Config(Config),
     Database(Database),
@@ -2760,6 +2818,22 @@ pub struct NetworkPolicy {
     pub value: NetworkPolicyValue,
 }
 
+#[derive(
+    Clone,
+    Debug,
+    PartialEq,
+    Eq,
+    PartialOrd,
+    Ord,
+    Serialize,
+    Deserialize,
+    Arbitrary
+)]
+pub struct ClusterReplicaSize {
+    pub key: ClusterReplicaSizeKey,
+    pub value: ClusterReplicaSizeValue,
+}
+
 #[derive(
     Clone,
     Debug,