sql: add CREATE/DROP CLUSTER REPLICA SIZE DDL

Add SQL DDL for creating and dropping user-defined cluster replica
sizes, gated behind the `enable_custom_cluster_replica_sizes` feature
flag (default: off).

Syntax:
  CREATE CLUSTER REPLICA SIZE <name> (
      CREDITS PER HOUR = '<numeric>',
      [WORKERS = <n>],
      [SCALE = <n>],
      [MEMORY LIMIT = <bytes>],
      [CPU LIMIT = <nanocpus>],
      [DISK LIMIT = <bytes>],
      [CPU EXCLUSIVE = <bool>],
      [DISABLED = <bool>]
  );
  DROP CLUSTER REPLICA SIZE <name>;

Key behaviors:
- Feature-flag gated: requires `enable_custom_cluster_replica_sizes`
- Cannot create a size with a name that already exists (builtin or user)
- Cannot drop builtin sizes (those synced from env var)
- Cannot drop a size that is in use by an existing cluster replica
- CREDITS PER HOUR is required; WORKERS defaults to 1, SCALE defaults to 1

Implementation follows the established DDL pattern (NetworkPolicy):
parser -> planner -> sequencer -> catalog transact, with exhaustive
match coverage across all statement/plan/response dispatchers.

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

Author: jubrad

src/adapter/src/catalog/apply.rs

@@ -1476,10 +1476,20 @@ impl CatalogState {
                         .expect("could not pack audit log update"),
                 ]
             }
+            StateUpdateKind::ClusterReplicaSize(ref size) => {
+                if size.allocation.disabled {
+                    Vec::new()
+                } else {
+                    vec![CatalogState::pack_replica_size_update(
+                        &size.name,
+                        &size.allocation,
+                        diff,
+                    )]
+                }
+            }
             StateUpdateKind::Database(_)
             | StateUpdateKind::Schema(_)
             | StateUpdateKind::NetworkPolicy(_)
-            | StateUpdateKind::ClusterReplicaSize(_)
             | StateUpdateKind::StorageCollectionMetadata(_)
             | StateUpdateKind::UnfinalizedShard(_) => Vec::new(),
         }

src/adapter/src/catalog/builtin_table_updates.rs

@@ -35,6 +35,7 @@ use mz_catalog::memory::objects::{
     CatalogEntry, CatalogItem, ClusterVariant, Connection, ContinualTask, DataSourceDesc, Func,
     Index, MaterializedView, Sink, Table, TableDataSource, Type, View,
 };
+use mz_controller::clusters::ReplicaAllocation;
 use mz_controller::clusters::{
     ManagedReplicaAvailabilityZones, ManagedReplicaLocation, ReplicaLocation,
 };
@@ -1933,34 +1934,36 @@ impl CatalogState {
                 continue;
             }
 
-            // Just invent something when the limits are `None`, which only happens in non-prod
-            // environments (tests, process orchestrator, etc.)
-            let cpu_limit = alloc.cpu_limit.unwrap_or(CpuLimit::MAX);
-            let MemoryLimit(ByteSize(memory_bytes)) =
-                (alloc.memory_limit).unwrap_or(MemoryLimit::MAX);
-            let DiskLimit(ByteSize(disk_bytes)) =
-                (alloc.disk_limit).unwrap_or(DiskLimit::ARBITRARY);
-
-            let row = Row::pack_slice(&[
-                size.as_str().into(),
-                u64::cast_from(alloc.scale).into(),
-                u64::cast_from(alloc.workers).into(),
-                cpu_limit.as_nanocpus().into(),
-                memory_bytes.into(),
-                disk_bytes.into(),
-                (alloc.credits_per_hour).into(),
-            ]);
-
-            updates.push(BuiltinTableUpdate::row(
-                &*MZ_CLUSTER_REPLICA_SIZES,
-                row,
-                Diff::ONE,
-            ));
+            updates.push(Self::pack_replica_size_update(size, alloc, Diff::ONE));
         }
 
         updates
     }
 
+    pub fn pack_replica_size_update(
+        size: &str,
+        alloc: &ReplicaAllocation,
+        diff: Diff,
+    ) -> BuiltinTableUpdate<&'static BuiltinTable> {
+        // Just invent something when the limits are `None`, which only happens in non-prod
+        // environments (tests, process orchestrator, etc.)
+        let cpu_limit = alloc.cpu_limit.unwrap_or(CpuLimit::MAX);
+        let MemoryLimit(ByteSize(memory_bytes)) = (alloc.memory_limit).unwrap_or(MemoryLimit::MAX);
+        let DiskLimit(ByteSize(disk_bytes)) = (alloc.disk_limit).unwrap_or(DiskLimit::ARBITRARY);
+
+        let row = Row::pack_slice(&[
+            size.into(),
+            u64::cast_from(alloc.scale).into(),
+            u64::cast_from(alloc.workers).into(),
+            cpu_limit.as_nanocpus().into(),
+            memory_bytes.into(),
+            disk_bytes.into(),
+            (alloc.credits_per_hour).into(),
+        ]);
+
+        BuiltinTableUpdate::row(&*MZ_CLUSTER_REPLICA_SIZES, row, diff)
+    }
+
     pub fn pack_subscribe_update(
         &self,
         id: GlobalId,

src/adapter/src/catalog/transact.rs

@@ -165,6 +165,23 @@ pub enum Op {
         name: String,
         owner_id: RoleId,
     },
+    CreateClusterReplicaSize {
+        name: String,
+        workers: usize,
+        scale: u16,
+        credits_per_hour: mz_repr::adt::numeric::Numeric,
+        memory_limit: Option<u64>,
+        cpu_limit: Option<u64>,
+        disk_limit: Option<u64>,
+        cpu_exclusive: bool,
+        disabled: bool,
+        selectors: BTreeMap<String, String>,
+        is_cc: bool,
+        swap_enabled: bool,
+    },
+    DropClusterReplicaSize {
+        name: String,
+    },
     Comment {
         object_id: CommentObjectId,
         sub_component: Option<usize>,
@@ -1664,6 +1681,64 @@ impl Catalog {
 
                 info!("created network policy {name} ({id})");
             }
+            Op::CreateClusterReplicaSize {
+                name,
+                workers,
+                scale,
+                credits_per_hour,
+                memory_limit,
+                cpu_limit,
+                disk_limit,
+                cpu_exclusive,
+                disabled,
+                selectors,
+                is_cc,
+                swap_enabled,
+            } => {
+                // Build ReplicaAllocation from the fields
+                use mz_orchestrator::{CpuLimit, DiskLimit, MemoryLimit};
+                use std::num::NonZero;
+
+                let allocation = mz_controller::clusters::ReplicaAllocation {
+                    memory_limit: memory_limit.map(|b| MemoryLimit(bytesize::ByteSize(b))),
+                    memory_request: None,
+                    cpu_limit: cpu_limit.map(|n| {
+                        CpuLimit::from_millicpus(usize::try_from(n / 1_000_000).expect("cpu fits"))
+                    }),
+                    cpu_request: None,
+                    disk_limit: disk_limit.map(|b| DiskLimit(bytesize::ByteSize(b))),
+                    scale: NonZero::new(scale).expect("scale must be non-zero"),
+                    workers: NonZero::new(workers).expect("workers must be non-zero"),
+                    credits_per_hour,
+                    cpu_exclusive,
+                    is_cc,
+                    swap_enabled,
+                    disabled,
+                    selectors,
+                };
+                tx.insert_cluster_replica_size(name, allocation, false)?;
+            }
+            Op::DropClusterReplicaSize { name } => {
+                // Check that it exists and is not builtin
+                let sizes: Vec<_> = tx.get_cluster_replica_sizes().collect();
+                let size = sizes.iter().find(|s| s.name == name);
+                match size {
+                    Some(s) if s.builtin => {
+                        return Err(AdapterError::Unstructured(anyhow::anyhow!(
+                            "cannot drop builtin cluster replica size '{}'",
+                            name
+                        )));
+                    }
+                    Some(_) => {
+                        tx.remove_cluster_replica_size(&name)?;
+                    }
+                    None => {
+                        return Err(AdapterError::PlanError(PlanError::Catalog(
+                            SqlCatalogError::UnknownClusterReplicaSize(name),
+                        )));
+                    }
+                }
+            }
             Op::Comment {
                 object_id,
                 sub_component,

src/adapter/src/command.rs

@@ -589,6 +589,10 @@ pub enum ExecuteResponse {
     CreatedType,
     /// The requested network policy was created.
     CreatedNetworkPolicy,
+    /// The requested cluster replica size was created.
+    CreatedClusterReplicaSize,
+    /// The requested cluster replica size was dropped.
+    DroppedClusterReplicaSize,
     /// The requested prepared statement was removed.
     Deallocate { all: bool },
     /// The requested cursor was declared.
@@ -755,6 +759,12 @@ impl TryInto<ExecuteResponse> for ExecuteResponseKind {
                 Ok(ExecuteResponse::CreatedMaterializedView)
             }
             ExecuteResponseKind::CreatedNetworkPolicy => Ok(ExecuteResponse::CreatedNetworkPolicy),
+            ExecuteResponseKind::CreatedClusterReplicaSize => {
+                Ok(ExecuteResponse::CreatedClusterReplicaSize)
+            }
+            ExecuteResponseKind::DroppedClusterReplicaSize => {
+                Ok(ExecuteResponse::DroppedClusterReplicaSize)
+            }
             ExecuteResponseKind::CreatedContinualTask => Ok(ExecuteResponse::CreatedContinualTask),
             ExecuteResponseKind::CreatedType => Ok(ExecuteResponse::CreatedType),
             ExecuteResponseKind::Deallocate => Err(()),
@@ -820,6 +830,8 @@ impl ExecuteResponse {
             CreatedContinualTask { .. } => Some("CREATE CONTINUAL TASK".into()),
             CreatedType => Some("CREATE TYPE".into()),
             CreatedNetworkPolicy => Some("CREATE NETWORKPOLICY".into()),
+            CreatedClusterReplicaSize => Some("CREATE CLUSTER REPLICA SIZE".into()),
+            DroppedClusterReplicaSize => Some("DROP CLUSTER REPLICA SIZE".into()),
             Deallocate { all } => Some(format!("DEALLOCATE{}", if *all { " ALL" } else { "" })),
             DeclaredCursor => Some("DECLARE CURSOR".into()),
             Deleted(n) => Some(format!("DELETE {}", n)),
@@ -914,6 +926,8 @@ impl ExecuteResponse {
             CreateType => &[CreatedType],
             PlanKind::Deallocate => &[ExecuteResponseKind::Deallocate],
             CreateNetworkPolicy => &[CreatedNetworkPolicy],
+            CreateClusterReplicaSize => &[CreatedClusterReplicaSize],
+            DropClusterReplicaSize => &[DroppedClusterReplicaSize],
             Declare => &[DeclaredCursor],
             DiscardTemp => &[DiscardedTemp],
             DiscardAll => &[DiscardedAll],

src/adapter/src/coord/appends.rs

@@ -957,6 +957,8 @@ pub(crate) fn waiting_on_startup_appends(
         | Plan::CreateSchema(_)
         | Plan::CreateRole(_)
         | Plan::CreateNetworkPolicy(_)
+        | Plan::CreateClusterReplicaSize(_)
+        | Plan::DropClusterReplicaSize(_)
         | Plan::CreateCluster(_)
         | Plan::CreateClusterReplica(_)
         | Plan::CreateContinualTask(_)

src/adapter/src/coord/catalog_serving.rs

@@ -82,6 +82,8 @@ pub fn auto_run_on_catalog_server<'a, 's, 'p>(
         | Plan::CreateSchema(_)
         | Plan::CreateRole(_)
         | Plan::CreateNetworkPolicy(_)
+        | Plan::CreateClusterReplicaSize(_)
+        | Plan::DropClusterReplicaSize(_)
         | Plan::CreateCluster(_)
         | Plan::CreateClusterReplica(_)
         | Plan::CreateContinualTask(_)

src/adapter/src/coord/command_handler.rs

@@ -1296,6 +1296,8 @@ impl Coordinator {
                     | Statement::CreateView(_)
                     | Statement::CreateWebhookSource(_)
                     | Statement::CreateNetworkPolicy(_)
+                    | Statement::CreateClusterReplicaSize(_)
+                    | Statement::DropClusterReplicaSize(_)
                     | Statement::Delete(_)
                     | Statement::DropObjects(_)
                     | Statement::DropOwned(_)

src/adapter/src/coord/ddl.rs

@@ -1291,7 +1291,9 @@ impl Coordinator {
                 | Op::ResetAllSystemConfiguration { .. }
                 | Op::Comment { .. }
                 | Op::WeirdStorageUsageUpdates { .. }
-                | Op::InjectAuditEvents { .. } => {}
+                | Op::InjectAuditEvents { .. }
+                | Op::CreateClusterReplicaSize { .. }
+                | Op::DropClusterReplicaSize { .. } => {}
             }
         }
 

src/adapter/src/coord/sequencer.rs

@@ -302,6 +302,18 @@ impl Coordinator {
                         .await;
                     ctx.retire(res);
                 }
+                Plan::CreateClusterReplicaSize(plan) => {
+                    let res = self
+                        .sequence_create_cluster_replica_size(ctx.session(), plan)
+                        .await;
+                    ctx.retire(res);
+                }
+                Plan::DropClusterReplicaSize(plan) => {
+                    let res = self
+                        .sequence_drop_cluster_replica_size(ctx.session(), plan)
+                        .await;
+                    ctx.retire(res);
+                }
                 Plan::Comment(plan) => {
                     let result = self.sequence_comment_on(ctx.session(), plan).await;
                     ctx.retire(result);

src/adapter/src/coord/sequencer/inner.rs

@@ -905,6 +905,54 @@ impl Coordinator {
             .map(|_| ExecuteResponse::CreatedNetworkPolicy)
     }
 
+    #[instrument]
+    pub(super) async fn sequence_create_cluster_replica_size(
+        &mut self,
+        _session: &Session,
+        plan: plan::CreateClusterReplicaSizePlan,
+    ) -> Result<ExecuteResponse, AdapterError> {
+        let op = catalog::Op::CreateClusterReplicaSize {
+            name: plan.name,
+            workers: plan.workers,
+            scale: plan.scale,
+            credits_per_hour: plan.credits_per_hour,
+            memory_limit: plan.memory_limit,
+            cpu_limit: plan.cpu_limit,
+            disk_limit: plan.disk_limit,
+            cpu_exclusive: plan.cpu_exclusive,
+            disabled: plan.disabled,
+            selectors: plan.selectors,
+            is_cc: plan.is_cc,
+            swap_enabled: plan.swap_enabled,
+        };
+        self.catalog_transact(None, vec![op])
+            .await
+            .map(|_| ExecuteResponse::CreatedClusterReplicaSize)
+    }
+
+    #[instrument]
+    pub(super) async fn sequence_drop_cluster_replica_size(
+        &mut self,
+        _session: &Session,
+        plan: plan::DropClusterReplicaSizePlan,
+    ) -> Result<ExecuteResponse, AdapterError> {
+        // Check in-use
+        if self
+            .catalog()
+            .state()
+            .is_cluster_replica_size_in_use(&plan.name)
+        {
+            return Err(AdapterError::Unstructured(anyhow::anyhow!(
+                "cannot drop cluster replica size '{}' because it is in use by an existing replica",
+                plan.name
+            )));
+        }
+        let op = catalog::Op::DropClusterReplicaSize { name: plan.name };
+        self.catalog_transact(None, vec![op])
+            .await
+            .map(|_| ExecuteResponse::DroppedClusterReplicaSize)
+    }
+
     #[instrument]
     pub(super) async fn sequence_alter_network_policy(
         &mut self,