sql: add CREATE/DROP CLUSTER REPLICA SIZE DDL

Add SQL DDL for creating and dropping user-defined cluster replica
sizes. Sizes are immutable once created; drop and recreate to change.

Syntax:
  CREATE CLUSTER REPLICA SIZE <name> (
      CREDITS PER HOUR = '<numeric>',  -- required
      WORKERS = <n>,                   -- default 1
      SCALE = <n>,                     -- default 1
      MEMORY LIMIT = '<size>',         -- e.g. '4GiB', '512MiB', '1GB'
      CPU LIMIT = '<cpu>',             -- e.g. '0.5' (cores), '500m' (millicpus)
      DISK LIMIT = '<size>',
      CPU EXCLUSIVE = <bool>,
      DISABLED = <bool>,
      IS CC = <bool>,                  -- default true
      SWAP ENABLED = <bool>,
      NODE SELECTORS = '<json>'        -- e.g. '{"kubernetes.io/arch": "arm64"}'
  );
  DROP CLUSTER REPLICA SIZE <name>;

Access control:
- Gated behind enable_custom_cluster_replica_sizes feature flag
- mz_system bypasses the feature flag (always allowed)
- RBAC requires superuser for both CREATE and DROP
- Cannot drop builtin sizes (from env var) or sizes in use by replicas
- Cannot create a size with a name that already exists

Human-readable units:
- Memory/disk: GiB, MiB, GB, MB, kB, or raw bytes
- CPU: cores (0.5), millicpus (500m), or raw nanocpus

Structured errors for drop rejection (ClusterReplicaSizeInUse,
ReadOnlyClusterReplicaSize). Audit log events for create/drop with
ObjectType::ClusterReplicaSize.

Tests:
- SLT end-to-end test covering feature flag gating, create, use with
  cluster, in-use drop rejection, builtin drop rejection, human-
  readable units (GiB, millicpus), and cleanup
- Parser roundtrip test
- Updated snapshot tests

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

Author: jubrad

src/adapter/src/catalog/apply.rs

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

src/adapter/src/catalog/builtin_table_updates.rs

@@ -16,17 +16,18 @@ use mz_audit_log::{EventDetails, EventType, ObjectType, VersionedEvent, Versione
 use mz_catalog::SYSTEM_CONN_ID;
 use mz_catalog::builtin::{
     BuiltinTable, MZ_AGGREGATES, MZ_ARRAY_TYPES, MZ_AUDIT_EVENTS, MZ_AWS_CONNECTIONS,
-    MZ_AWS_PRIVATELINK_CONNECTIONS, MZ_BASE_TYPES, MZ_CLUSTER_REPLICA_SIZES, MZ_CLUSTER_REPLICAS,
-    MZ_CLUSTER_SCHEDULES, MZ_CLUSTERS, MZ_COLUMNS, MZ_COMMENTS, MZ_CONNECTIONS, MZ_CONTINUAL_TASKS,
-    MZ_DEFAULT_PRIVILEGES, MZ_EGRESS_IPS, MZ_FUNCTIONS, MZ_HISTORY_RETENTION_STRATEGIES,
-    MZ_ICEBERG_SINKS, MZ_INDEX_COLUMNS, MZ_INDEXES, MZ_KAFKA_CONNECTIONS, MZ_KAFKA_SINKS,
-    MZ_KAFKA_SOURCE_TABLES, MZ_KAFKA_SOURCES, MZ_LICENSE_KEYS, MZ_LIST_TYPES, MZ_MAP_TYPES,
-    MZ_MATERIALIZED_VIEW_REFRESH_STRATEGIES, MZ_MYSQL_SOURCE_TABLES, MZ_OBJECT_DEPENDENCIES,
-    MZ_OBJECT_GLOBAL_IDS, MZ_OPERATORS, MZ_POSTGRES_SOURCE_TABLES, MZ_POSTGRES_SOURCES,
-    MZ_PSEUDO_TYPES, MZ_REPLACEMENTS, MZ_ROLE_AUTH, MZ_ROLE_PARAMETERS, MZ_ROLES, MZ_SECRETS,
-    MZ_SESSIONS, MZ_SINKS, MZ_SOURCE_REFERENCES, MZ_SOURCES, MZ_SQL_SERVER_SOURCE_TABLES,
-    MZ_SSH_TUNNEL_CONNECTIONS, MZ_STORAGE_USAGE_BY_SHARD, MZ_SUBSCRIPTIONS, MZ_SYSTEM_PRIVILEGES,
-    MZ_TABLES, MZ_TYPE_PG_METADATA, MZ_TYPES, MZ_VIEWS, MZ_WEBHOOKS_SOURCES,
+    MZ_AWS_PRIVATELINK_CONNECTIONS, MZ_BASE_TYPES, MZ_CLUSTER_REPLICA_SIZE_DETAILS,
+    MZ_CLUSTER_REPLICA_SIZES, MZ_CLUSTER_REPLICAS, MZ_CLUSTER_SCHEDULES, MZ_CLUSTERS, MZ_COLUMNS,
+    MZ_COMMENTS, MZ_CONNECTIONS, MZ_CONTINUAL_TASKS, MZ_DEFAULT_PRIVILEGES, MZ_EGRESS_IPS,
+    MZ_FUNCTIONS, MZ_HISTORY_RETENTION_STRATEGIES, MZ_ICEBERG_SINKS, MZ_INDEX_COLUMNS, MZ_INDEXES,
+    MZ_KAFKA_CONNECTIONS, MZ_KAFKA_SINKS, MZ_KAFKA_SOURCE_TABLES, MZ_KAFKA_SOURCES,
+    MZ_LICENSE_KEYS, MZ_LIST_TYPES, MZ_MAP_TYPES, MZ_MATERIALIZED_VIEW_REFRESH_STRATEGIES,
+    MZ_MYSQL_SOURCE_TABLES, MZ_OBJECT_DEPENDENCIES, MZ_OBJECT_GLOBAL_IDS, MZ_OPERATORS,
+    MZ_POSTGRES_SOURCE_TABLES, MZ_POSTGRES_SOURCES, MZ_PSEUDO_TYPES, MZ_REPLACEMENTS, MZ_ROLE_AUTH,
+    MZ_ROLE_PARAMETERS, MZ_ROLES, MZ_SECRETS, MZ_SESSIONS, MZ_SINKS, MZ_SOURCE_REFERENCES,
+    MZ_SOURCES, MZ_SQL_SERVER_SOURCE_TABLES, MZ_SSH_TUNNEL_CONNECTIONS, MZ_STORAGE_USAGE_BY_SHARD,
+    MZ_SUBSCRIPTIONS, MZ_SYSTEM_PRIVILEGES, MZ_TABLES, MZ_TYPE_PG_METADATA, MZ_TYPES, MZ_VIEWS,
+    MZ_WEBHOOKS_SOURCES,
 };
 use mz_catalog::config::AwsPrincipalContext;
 use mz_catalog::durable::SourceReferences;
@@ -35,6 +36,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 +1935,67 @@ 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,
+            // The builtin flag isn't available from ClusterReplicaSizeMap; the details
+            // table is populated correctly via state updates in apply.rs which has
+            // access to the full ClusterReplicaSize with the builtin flag.
+            updates.extend(Self::pack_replica_size_update(
+                size,
+                alloc,
+                false,
                 Diff::ONE,
             ));
         }
 
         updates
     }
 
+    pub fn pack_replica_size_update(
+        size: &str,
+        alloc: &ReplicaAllocation,
+        builtin: bool,
+        diff: Diff,
+    ) -> Vec<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 public_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(),
+        ]);
+
+        let selectors_value =
+            serde_json::to_value(&alloc.selectors).expect("selectors serializable");
+        let selectors_jsonb = Jsonb::from_serde_json(selectors_value).expect("valid jsonb");
+        let details_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(),
+            alloc.cpu_exclusive.into(),
+            alloc.is_cc.into(),
+            alloc.swap_enabled.into(),
+            alloc.disabled.into(),
+            builtin.into(),
+            selectors_jsonb.into_row().into_element(),
+        ]);
+
+        vec![
+            BuiltinTableUpdate::row(&*MZ_CLUSTER_REPLICA_SIZES, public_row, diff),
+            BuiltinTableUpdate::row(&*MZ_CLUSTER_REPLICA_SIZE_DETAILS, details_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,91 @@ 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.clone(), allocation, false)?;
+
+                CatalogState::add_to_audit_log(
+                    &state.system_configuration,
+                    oracle_write_ts,
+                    session,
+                    tx,
+                    audit_events,
+                    EventType::Create,
+                    mz_audit_log::ObjectType::ClusterReplicaSize,
+                    EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
+                        id: name.clone(),
+                        name: name.clone(),
+                    }),
+                )?;
+            }
+            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::Catalog(Error::new(
+                            ErrorKind::ReadOnlyClusterReplicaSize(name),
+                        )));
+                    }
+                    Some(_) => {
+                        tx.remove_cluster_replica_size(&name)?;
+                    }
+                    None => {
+                        return Err(AdapterError::PlanError(PlanError::Catalog(
+                            SqlCatalogError::UnknownClusterReplicaSize(name),
+                        )));
+                    }
+                }
+
+                CatalogState::add_to_audit_log(
+                    &state.system_configuration,
+                    oracle_write_ts,
+                    session,
+                    tx,
+                    audit_events,
+                    EventType::Drop,
+                    mz_audit_log::ObjectType::ClusterReplicaSize,
+                    EventDetails::IdNameV1(mz_audit_log::IdNameV1 {
+                        id: name.clone(),
+                        name: name.clone(),
+                    }),
+                )?;
+            }
             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);