row-spine: per-column arrangement compression (alpha, default off)

misc/python/materialize/mzcompose/__init__.py

@@ -71,6 +71,7 @@ def get_minimal_system_parameters(
             "true" if version >= MzVersion.parse_mz("v0.132.0-dev") else "false"
         ),
         "enable_alter_swap": "true",
+        "enable_arrangement_column_compression_alpha": "false",
         "enable_arrangement_dictionary_compression_alpha": "false",
         "enable_case_literal_transform": "true",
         "enable_cast_elimination": "true",

misc/python/materialize/parallel_workload/action.py

@@ -1606,6 +1606,9 @@ def __init__(
         self.flags_with_values["enable_arrangement_dictionary_compression_alpha"] = (
             BOOLEAN_FLAG_VALUES
         )
+        self.flags_with_values["enable_arrangement_column_compression_alpha"] = (
+            BOOLEAN_FLAG_VALUES
+        )
         self.flags_with_values["enable_compute_peek_response_stash"] = (
             BOOLEAN_FLAG_VALUES
         )

src/compute-types/src/dyncfgs.rs

@@ -399,6 +399,18 @@ pub const ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA: Config<bool> = Config
     "Enable arrangement dictionary compression (alpha; not yet production-ready).",
 );
 
+/// Whether to enable per-column arrangement compression (the `RowCodec`
+/// framework). Independent of, and not combined with, dictionary compression.
+///
+/// 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
 /// `compute_peek_response_stash_threshold_bytes`.
@@ -525,6 +537,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&COMPUTE_LOGICAL_BACKPRESSURE_MAX_RETAINED_CAPABILITIES)
         .add(&COMPUTE_LOGICAL_BACKPRESSURE_INFLIGHT_SLACK)
         .add(&ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA)
+        .add(&ENABLE_ARRANGEMENT_COLUMN_COMPRESSION_ALPHA)
         .add(&ENABLE_PEEK_RESPONSE_STASH)
         .add(&PEEK_RESPONSE_STASH_THRESHOLD_BYTES)
         .add(&PEEK_RESPONSE_STASH_BATCH_MAX_RUNS)

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,6 +499,8 @@ impl<'a> ActiveComputeState<'a> {
             config.arrangement_dictionary_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);