Scoped feature flags 1/3: scope declaration and per-replica override plumbing (#37079)

### Motivation

First of a three-PR stack splitting #36959 (scoped feature flags) for
review:
**1/3 (this) → 2/3 #37080 → 3/3 #36959**. Design:
doc/developer/design/20260609_scoped_feature_flags.md (#36947).

This PR is the additive foundation: it introduces the vocabulary the
later PRs
build on, with no behavior change on its own.

### Description

* `ParameterScope` (`Environment` / `Cluster` / `Replica`), declared on
system
vars and dyncfg `Config`s and carried through to the synced system vars.
The
declaration is the single source of truth for which contexts get
evaluated.
* The size-family taxonomy: `ReplicaAllocation::family` plus a size-map
`family`
  field, with a `cc` / `legacy` fallback for sizes that don't set one.
* The compute controller's per-replica dyncfg override layer
(`update_replica_dyncfg_overrides` + per-replica command
specialization),
  inert until the adapter wires it in 3/3.

Nothing consumes the scope or the override layer yet, so this is a
no-op.

### Verification

`test_replica_allocation_family` covers the size→family fallback; the
rest is
exercised by the later PRs in the stack.

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: antiguru

src/catalog/src/config.rs

@@ -189,6 +189,7 @@ impl ClusterReplicaSizeMap {
                             credits_per_hour: 1.into(),
                             cpu_exclusive: false,
                             is_cc: false,
+                            family: None,
                             swap_enabled: false,
                             disabled: false,
                             selectors: BTreeMap::default(),
@@ -212,6 +213,7 @@ impl ClusterReplicaSizeMap {
                     credits_per_hour: scale.into(),
                     cpu_exclusive: false,
                     is_cc: false,
+                    family: None,
                     swap_enabled: false,
                     disabled: false,
                     selectors: BTreeMap::default(),
@@ -230,6 +232,7 @@ impl ClusterReplicaSizeMap {
                     credits_per_hour: scale.into(),
                     cpu_exclusive: false,
                     is_cc: false,
+                    family: None,
                     swap_enabled: false,
                     disabled: false,
                     selectors: BTreeMap::default(),
@@ -248,6 +251,7 @@ impl ClusterReplicaSizeMap {
                     credits_per_hour: 1.into(),
                     cpu_exclusive: false,
                     is_cc: false,
+                    family: None,
                     swap_enabled: false,
                     disabled: false,
                     selectors: BTreeMap::default(),
@@ -267,6 +271,7 @@ impl ClusterReplicaSizeMap {
                 credits_per_hour: 2.into(),
                 cpu_exclusive: false,
                 is_cc: false,
+                family: None,
                 swap_enabled: false,
                 disabled: false,
                 selectors: BTreeMap::default(),
@@ -285,6 +290,7 @@ impl ClusterReplicaSizeMap {
                 credits_per_hour: 0.into(),
                 cpu_exclusive: false,
                 is_cc: true,
+                family: None,
                 swap_enabled: false,
                 disabled: true,
                 selectors: BTreeMap::default(),

src/compute-client/src/controller.rs

@@ -46,7 +46,7 @@ use mz_compute_types::dataflows::DataflowDescription;
 use mz_compute_types::dyncfgs::{
     COMPUTE_REPLICA_EXPIRATION_OFFSET, ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA,
 };
-use mz_dyncfg::ConfigSet;
+use mz_dyncfg::{ConfigSet, ConfigUpdates};
 use mz_expr::RowSetFinishing;
 use mz_expr::row::RowCollection;
 use mz_ore::cast::CastFrom;
@@ -633,6 +633,24 @@ impl ComputeController {
         self.config.update(config_params);
     }
 
+    /// Replaces the per-replica dyncfg overrides for the given instances.
+    ///
+    /// This only stores the overrides; callers should follow with a
+    /// configuration push (e.g. [`Self::update_configuration`]) so existing
+    /// replicas observe the new values. Instances absent from `overrides` have
+    /// their overrides cleared, so a replica that no longer has an override
+    /// reverts to the environment-wide configuration. Used by the scoped
+    /// feature flags (replica-local) layer.
+    pub fn update_replica_dyncfg_overrides(
+        &mut self,
+        mut overrides: BTreeMap<ComputeInstanceId, BTreeMap<ReplicaId, ConfigUpdates>>,
+    ) {
+        for (id, instance) in self.instances.iter_mut() {
+            let instance_overrides = overrides.remove(id).unwrap_or_default();
+            instance.call(move |i| i.update_replica_dyncfg_overrides(instance_overrides));
+        }
+    }
+
     /// Mark the end of any initialization commands.
     ///
     /// The implementor may wait for this method to be called before implementing prior commands,

src/compute-client/src/controller/instance.rs

@@ -26,7 +26,7 @@ use mz_compute_types::sources::SourceInstanceDesc;
 use mz_controller_types::dyncfgs::{
     ENABLE_PAUSED_CLUSTER_READHOLD_DOWNGRADE, WALLCLOCK_LAG_RECORDING_INTERVAL,
 };
-use mz_dyncfg::ConfigSet;
+use mz_dyncfg::{ConfigSet, ConfigUpdates};
 use mz_expr::RowSetFinishing;
 use mz_ore::cast::CastFrom;
 use mz_ore::channel::instrumented_unbounded_channel;
@@ -140,6 +140,13 @@ pub(super) struct Instance {
     workload_class: Option<String>,
     /// The replicas of this compute instance.
     replicas: BTreeMap<ReplicaId, ReplicaState>,
+    /// Per-replica dyncfg overrides, merged into the `UpdateConfiguration`
+    /// command sent to each replica (and into the command-history replay used
+    /// to hydrate new replicas). Populated from the scoped feature flags
+    /// (replica-local) layer; empty by default, in which case every replica
+    /// receives the unmodified environment-wide configuration. Stores only the
+    /// values that differ from the environment-wide value, so the map is sparse.
+    replica_dyncfg_overrides: BTreeMap<ReplicaId, ConfigUpdates>,
     /// Currently installed compute collections.
     ///
     /// New entries are added for all collections exported from dataflows created through
@@ -845,6 +852,7 @@ impl Instance {
             read_only,
             workload_class,
             replicas,
+            replica_dyncfg_overrides: _,
             collections,
             log_sources: _,
             peeks,
@@ -950,6 +958,7 @@ impl Instance {
             read_only,
             workload_class: None,
             replicas: Default::default(),
+            replica_dyncfg_overrides: Default::default(),
             collections,
             log_sources,
             peeks: Default::default(),
@@ -1091,21 +1100,57 @@ impl Instance {
     #[mz_ore::instrument(level = "debug")]
     fn send(&mut self, cmd: ComputeCommand) {
         // Record the command so that new replicas can be brought up to speed.
+        // We record the *base* (un-specialized) command, so that the per-replica
+        // dyncfg overrides are re-applied at replay time in `add_replica` rather
+        // than baked into the shared history.
         self.history.push(cmd.clone());
 
         let target_replica = self.target_replica(&cmd);
 
+        // Borrow the overrides separately from `self.replicas` so the per-replica
+        // specialization below does not conflict with the mutable replica borrow.
+        let overrides = &self.replica_dyncfg_overrides;
+
         if let Some(rid) = target_replica {
             if let Some(replica) = self.replicas.get_mut(&rid) {
+                let cmd = Self::specialize_command_for_replica(cmd, rid, overrides);
                 let _ = replica.client.send(cmd);
             }
         } else {
-            for replica in self.replicas.values_mut() {
-                let _ = replica.client.send(cmd.clone());
+            for (rid, replica) in self.replicas.iter_mut() {
+                let cmd = Self::specialize_command_for_replica(cmd.clone(), *rid, overrides);
+                let _ = replica.client.send(cmd);
             }
         }
     }
 
+    /// Specializes a command for a specific replica by merging that replica's
+    /// dyncfg override (if any) into an `UpdateConfiguration` command. All other
+    /// commands, and replicas without an override, are returned unchanged.
+    fn specialize_command_for_replica(
+        mut cmd: ComputeCommand,
+        replica_id: ReplicaId,
+        overrides: &BTreeMap<ReplicaId, ConfigUpdates>,
+    ) -> ComputeCommand {
+        if let ComputeCommand::UpdateConfiguration(params) = &mut cmd
+            && let Some(over) = overrides.get(&replica_id)
+            && !over.updates.is_empty()
+        {
+            params.dyncfg_updates.extend(over.clone());
+        }
+        cmd
+    }
+
+    /// Replaces the per-replica dyncfg overrides. Callers should follow this
+    /// with a configuration push (e.g. `update_configuration`) so that existing
+    /// replicas observe the new overrides.
+    pub(super) fn update_replica_dyncfg_overrides(
+        &mut self,
+        overrides: BTreeMap<ReplicaId, ConfigUpdates>,
+    ) {
+        self.replica_dyncfg_overrides = overrides;
+    }
+
     /// Determine the target replica for a compute command. Retrieves the
     /// collection named by the command, and returns the target replica if
     /// it is set, and None if not set, or the command doesn't name a collection.
@@ -1183,7 +1228,13 @@ impl Instance {
                 continue;
             }
 
-            if client.send(command.clone()).is_err() {
+            // Re-apply this replica's dyncfg override to replayed config commands.
+            let command = Self::specialize_command_for_replica(
+                command.clone(),
+                id,
+                &self.replica_dyncfg_overrides,
+            );
+            if client.send(command).is_err() {
                 // We swallow the error here. On the next send, we will fail again, and
                 // restart the connection as well as this rehydration.
                 tracing::warn!("Replica {:?} connection terminated during hydration", id);

src/compute-types/src/dyncfgs.rs

@@ -11,7 +11,7 @@
 
 use std::time::Duration;
 
-use mz_dyncfg::{Config, ConfigSet};
+use mz_dyncfg::{Config, ConfigSet, ParameterScope};
 
 /// Whether rendering should use `half_join2` rather than DD's `half_join` for delta joins.
 ///
@@ -41,7 +41,8 @@ pub const ENABLE_COLUMN_PAGED_BATCHER: Config<bool> = Config::new(
     false,
     "Use the columnar-native paged merge batcher at arrange sites. When `false` (default), \
      arranges fall back to the legacy columnation `Col2ValBatcher` / `RowRowBuilder` path.",
-);
+)
+.scoped(ParameterScope::Replica);
 
 /// Allow the column-paged batcher's pager to actually evict chunks
 /// under memory pressure. Only meaningful when
@@ -56,7 +57,8 @@ pub const ENABLE_COLUMN_PAGED_BATCHER_SPILL: Config<bool> = Config::new(
     false,
     "Allow the column-paged batcher's pager to evict chunks under memory pressure. Only \
      meaningful when `enable_column_paged_batcher = true`.",
-);
+)
+.scoped(ParameterScope::Replica);
 
 /// Total resident-byte budget the column-paged batcher's tiered policy
 /// (`mz_timely_util::column_pager::policy::TieredPolicy`) is allowed to
@@ -91,7 +93,8 @@ pub const COLUMN_PAGED_BATCHER_LZ4: Config<bool> = Config::new(
     false,
     "Compress column-paged batcher chunks with lz4 on the spill path. Only meaningful when \
      `enable_column_paged_batcher_spill = true`.",
-);
+)
+.scoped(ParameterScope::Replica);
 
 /// Proactively evict the column-paged batcher's lz4-compressed spill chunks
 /// from RSS via `MADV_PAGEOUT` when spilling to the swap backend. Only
@@ -176,7 +179,8 @@ pub const LINEAR_JOIN_YIELDING: Config<&str> = Config::new(
 );
 
 /// Enable lgalloc.
-pub const ENABLE_LGALLOC: Config<bool> = Config::new("enable_lgalloc", true, "Enable lgalloc.");
+pub const ENABLE_LGALLOC: Config<bool> =
+    Config::new("enable_lgalloc", true, "Enable lgalloc.").scoped(ParameterScope::Replica);
 
 /// Enable lgalloc's eager memory return/reclamation feature.
 pub const ENABLE_LGALLOC_EAGER_RECLAMATION: Config<bool> = Config::new(

src/controller/src/clusters.rs

@@ -95,6 +95,16 @@ pub struct ReplicaAllocation {
     /// T-shirt size.
     #[serde(default = "default_true")]
     pub is_cc: bool,
+    /// The size *family* this size belongs to, e.g. the size `D.1-xsmall`
+    /// belongs to family `D` and the legacy t-shirt sizes belong to family
+    /// `legacy`. The family is the coarse axis and is *not* a prefix of the size
+    /// name in general. Used as the
+    /// `replica_size_family` attribute when evaluating replica-local scoped
+    /// feature flags (see the scoped feature flags design). When unset, the
+    /// family falls back to a value derived from [`Self::is_cc`] via
+    /// [`ReplicaAllocation::family`].
+    #[serde(default)]
+    pub family: Option<String>,
     /// Whether instances of this type use swap as the spill-to-disk mechanism.
     #[serde(default)]
     pub swap_enabled: bool,
@@ -106,6 +116,24 @@ pub struct ReplicaAllocation {
     pub selectors: BTreeMap<String, String>,
 }
 
+impl ReplicaAllocation {
+    /// The name of the size family this allocation belongs to, used as the
+    /// `replica_size_family` attribute when evaluating replica-local scoped
+    /// feature flags.
+    ///
+    /// Falls back to a value derived from [`Self::is_cc`] when [`Self::family`]
+    /// is unset: `"cc"` for modern sizes and `"legacy"` for the legacy t-shirt
+    /// sizes. This keeps the legacy family targetable even before every size
+    /// gains an explicit `family` in the size configuration.
+    pub fn family(&self) -> &str {
+        match &self.family {
+            Some(family) => family.as_str(),
+            None if self.is_cc => "cc",
+            None => "legacy",
+        }
+    }
+}
+
 fn default_true() -> bool {
     true
 }
@@ -146,6 +174,7 @@ fn test_replica_allocation_deserialization() {
             cpu_request: None,
             cpu_exclusive: false,
             is_cc: true,
+            family: None,
             swap_enabled: true,
             scale: NonZero::new(16).unwrap(),
             workers: NonZero::new(1).unwrap(),
@@ -182,6 +211,7 @@ fn test_replica_allocation_deserialization() {
             cpu_request: None,
             cpu_exclusive: true,
             is_cc: true,
+            family: None,
             swap_enabled: false,
             scale: NonZero::new(1).unwrap(),
             workers: NonZero::new(1).unwrap(),
@@ -198,6 +228,40 @@ fn test_replica_allocation_deserialization() {
     assert_ok!(serde_json::from_str::<ReplicaAllocation>(data));
 }
 
+#[mz_ore::test]
+#[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
+fn test_replica_allocation_family() {
+    let parse = |json: &str| -> ReplicaAllocation {
+        serde_json::from_str(json).expect("deserialization from JSON succeeds")
+    };
+
+    // An explicit `family` is used verbatim.
+    assert_eq!(
+        parse(r#"{"scale": 1, "workers": 1, "credits_per_hour": "0", "family": "D"}"#).family(),
+        "D"
+    );
+    // Without an explicit `family`, modern (`is_cc`) sizes fall back to "cc".
+    // `is_cc` defaults to true.
+    assert_eq!(
+        parse(r#"{"scale": 1, "workers": 1, "credits_per_hour": "0"}"#).family(),
+        "cc"
+    );
+    // Without an explicit `family`, legacy (non-`is_cc`) sizes fall back to
+    // "legacy".
+    assert_eq!(
+        parse(r#"{"scale": 1, "workers": 1, "credits_per_hour": "0", "is_cc": false}"#).family(),
+        "legacy"
+    );
+    // An explicit family wins even for a legacy size.
+    assert_eq!(
+        parse(
+            r#"{"scale": 1, "workers": 1, "credits_per_hour": "0", "is_cc": false, "family": "legacy-special"}"#
+        )
+        .family(),
+        "legacy-special"
+    );
+}
+
 /// Configures the location of a cluster replica.
 #[derive(Clone, Debug, Serialize, PartialEq)]
 pub enum ReplicaLocation {