compute: make per-column compression flag replica-scoped

Addresses review feedback on MaterializeInc#37111: scope
`enable_arrangement_column_compression_alpha` to the replica
(`ParameterScope::Replica`) so it can carry per-replica / per-size-family
overrides, like the other replica-local physical flags (lgalloc, persist pager,
column-paged batcher).

This also replaces the manual capture chain the flag previously borrowed from
dictionary compression — InstanceConfig/ReplicaConfig fields, the controller's
per-replica capture, and the `handle_create_instance` store — with the
replica-scoped idiom: the per-replica-resolved value arrives in the replica's
`worker_config`, and `apply_worker_config` mirrors it into the process-global
`mz_row_spine::COLUMN_COMPRESSION` flag (next to the lgalloc-region store).
The controller-level capture read only the environment value and would have
missed per-replica overrides. Re-applying on each config tick is safe: every
batch records its own codec, so a flip only affects batches sealed afterwards.

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

Author: frankmcsherry

src/compute-client/src/controller.rs

@@ -44,8 +44,7 @@ use mz_compute_types::ComputeInstanceId;
 use mz_compute_types::config::ComputeReplicaConfig;
 use mz_compute_types::dataflows::DataflowDescription;
 use mz_compute_types::dyncfgs::{
-    COMPUTE_REPLICA_EXPIRATION_OFFSET, ENABLE_ARRANGEMENT_COLUMN_COMPRESSION_ALPHA,
-    ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA,
+    COMPUTE_REPLICA_EXPIRATION_OFFSET, ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA,
 };
 use mz_dyncfg::{ConfigSet, ConfigUpdates};
 use mz_expr::RowSetFinishing;
@@ -721,9 +720,6 @@ impl ComputeController {
         // why a later flip of the flag only affects replicas created afterwards.
         let arrangement_dictionary_compression =
             ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA.get(&self.dyncfg);
-        // Likewise captured once at replica creation, independent of dictionary compression.
-        let arrangement_column_compression =
-            ENABLE_ARRANGEMENT_COLUMN_COMPRESSION_ALPHA.get(&self.dyncfg);
 
         let replica_config = ReplicaConfig {
             location,
@@ -736,7 +732,6 @@ impl ComputeController {
             grpc_client: self.config.grpc_client.clone(),
             expiration_offset: (!expiration_offset.is_zero()).then_some(expiration_offset),
             arrangement_dictionary_compression,
-            arrangement_column_compression,
         };
 
         let instance = self.instance_mut(instance_id).expect("validated");

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

@@ -993,7 +993,6 @@ impl Instance {
             logging: Default::default(),
             expiration_offset: Default::default(),
             arrangement_dictionary_compression: Default::default(),
-            arrangement_column_compression: Default::default(),
         };
 
         self.send(ComputeCommand::CreateInstance(Box::new(instance_config)));

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

@@ -51,8 +51,6 @@ pub(super) struct ReplicaConfig {
     pub expiration_offset: Option<Duration>,
     /// Whether arrangements on this replica use dictionary compression, captured at creation.
     pub arrangement_dictionary_compression: bool,
-    /// Whether arrangements on this replica use per-column compression, captured at creation.
-    pub arrangement_column_compression: bool,
 }
 
 /// A client for a replica task.
@@ -280,7 +278,6 @@ impl ReplicaTask {
                 }
                 config.arrangement_dictionary_compression =
                     self.config.arrangement_dictionary_compression;
-                config.arrangement_column_compression = self.config.arrangement_column_compression;
             }
             _ => {}
         }

src/compute-client/src/protocol/command.rs

@@ -281,13 +281,6 @@ pub struct InstanceConfig {
     /// held fixed for the replica's lifetime, so flipping the flag only affects replicas created
     /// afterwards rather than retroactively changing arrangements across the environment.
     pub arrangement_dictionary_compression: bool,
-    /// Whether arrangements created by this replica use per-column compression.
-    ///
-    /// Captured from `enable_arrangement_column_compression_alpha` when the replica is created and
-    /// held fixed for the replica's lifetime, exactly like `arrangement_dictionary_compression`.
-    /// The two are independent; per-column compression layers over raw row bytes, not over
-    /// dictionary-compressed bytes.
-    pub arrangement_column_compression: bool,
 }
 
 impl InstanceConfig {
@@ -310,14 +303,12 @@ impl InstanceConfig {
             peek_stash_persist_location: self_peek_stash_persist_location,
             // Captured at replica creation; intentionally not part of compatibility (see above).
             arrangement_dictionary_compression: _,
-            arrangement_column_compression: _,
         } = self;
         let InstanceConfig {
             logging: other_logging,
             expiration_offset: other_offset,
             peek_stash_persist_location: other_peek_stash_persist_location,
             arrangement_dictionary_compression: _,
-            arrangement_column_compression: _,
         } = other;
 
         // Logging is compatible if exactly the same.

src/compute-types/src/dyncfgs.rs

@@ -401,12 +401,15 @@ pub const ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA: Config<bool> = Config
 
 /// Whether to enable per-column arrangement compression (the `RowCodec`
 /// framework). Independent of, and not combined with, dictionary compression.
-/// Like dictionary compression, captured at replica creation and held fixed.
+///
+/// Replica-scoped: the per-replica-resolved value is applied to the process-global
+/// `mz_row_spine::COLUMN_COMPRESSION` flag in the compute replica's `apply_worker_config`.
 pub const ENABLE_ARRANGEMENT_COLUMN_COMPRESSION_ALPHA: Config<bool> = Config::new(
     "enable_arrangement_column_compression_alpha",
     false,
     "Enable per-column arrangement compression (alpha; not yet production-ready).",
-);
+)
+.scoped(ParameterScope::Replica);
 
 /// Whether to enable the peek response stash, for sending back large peek
 /// responses. The response stash will only be used for results that exceed

src/compute/src/compute_state.rs

@@ -308,6 +308,16 @@ impl ComputeState {
             std::sync::atomic::Ordering::Relaxed,
         );
 
+        // Per-column arrangement compression is replica-scoped (see
+        // `ENABLE_ARRANGEMENT_COLUMN_COMPRESSION_ALPHA`), so its per-replica-resolved value
+        // arrives in `worker_config`; mirror it into the process-global flag row-spine reads at
+        // seal time. Safe to re-apply on every tick: each batch records its own codec, so a flip
+        // only affects batches sealed afterwards.
+        mz_row_spine::COLUMN_COMPRESSION.store(
+            ENABLE_ARRANGEMENT_COLUMN_COMPRESSION_ALPHA.get(config),
+            std::sync::atomic::Ordering::Relaxed,
+        );
+
         // NB: arrangement dictionary compression is deliberately NOT applied here. Unlike the
         // settings above, it is captured once at replica creation (see `handle_create_instance`
         // and `InstanceConfig::arrangement_dictionary_compression`) and held fixed, so that
@@ -489,11 +499,8 @@ impl<'a> ActiveComputeState<'a> {
             config.arrangement_dictionary_compression,
             std::sync::atomic::Ordering::Relaxed,
         );
-        // Same one-shot treatment for per-column compression; see the note above.
-        mz_row_spine::COLUMN_COMPRESSION.store(
-            config.arrangement_column_compression,
-            std::sync::atomic::Ordering::Relaxed,
-        );
+        // NB: per-column compression is replica-scoped and applied in `apply_worker_config`
+        // (called just above), not captured on `InstanceConfig`.
 
         if let Some(offset) = config.expiration_offset {
             self.compute_state.apply_expiration_offset(offset);