adapter: fix graceful cluster reconfig with arrangement compresssion (#38241)

## The bug

Arrangement compression is a replica config shape dimension. It is part
of
`ReconfigurationTarget`, the reconcile kernel distinguishes replicas by
it, and
the controller transitions it through the hydrate-overlap exactly like
`SIZE`.
Three hand-maintained dimension lists forgot it.

**1. The realized-config leak (the one that actually hurts).** The
reshape path
writes the reconfiguration record and resets the shape carried by
`new_config`
back to the realized values, so the transition is deferred to the
record's
cut-over. That reset enumerated size, replication factor, availability
zones and
logging, and missed compression. So a compression-carrying `ALTER`
flipped the
realized config *immediately*. The baseline strategy then saw the new
shape,
decided the existing baseline replicas no longer matched, and bounced
them in the
middle of the reconfiguration. That defeats the graceful overlap the
record
exists to provide, on a statement the user expected to be graceful.

The fix stops hand-listing dimensions: the reset goes through
`realized_reconfiguration_target` and a new inverse
`apply_reconfiguration_target`. Both destructure `ReconfigurationTarget`
exhaustively, so a future dimension fails to compile until it is handled
here.

**2. The planner's `WAIT` gate** rejected `WITH (WAIT ...)` on a
compression-only
`ALTER` as having nothing to wait on, even though it transitions through
a
hydrate-overlap like any other shape change.

**3. Observability.** `mz_cluster_reconfigurations.changes` diffed only
the other
four dimensions, so a compression-only record reported an empty diff
while in
progress, and the `SHOW CLUSTERS` activity summary showed such a cluster
as
steady.

## Testing

The realized-config behaviour is asserted in `cluster-controller.td`,
where
testdrive's retries make it deterministic. The `cc_preserve` section now
checks
that a folding compression `ALTER` lands in the record's target and
`changes`
while the realized config stays untouched and only the overlap replica
is
bounced. A new section runs a compression-only `ALTER` through the
record path
end to end.

The `managed_cluster.slt` compression `ALTER`s previously depended on
the leak:
their `SHOW CREATE` readback only saw the new value because the reshape
wrote it
early. With the leak gone the value moves at cut-over, and the suite
pins
`enable_background_alter_cluster` on, so the statement returns before
that. Those
`ALTER`s are exercised for acceptance instead, one with a `WAIT` clause
to cover
the planner gate.

## Migration

Changing the MV definition changes its builtin fingerprint, so the
builtin schema
migration registers a replacement step for
`mz_cluster_reconfigurations`. The
catalog-docs golden and the `mz_show_clusters` plan golden are
rewritten.

Author: aljoscha

doc/user/content/reference/system-catalog/mz_internal.md

@@ -181,7 +181,7 @@ shape is in [`mz_clusters`](../mz_catalog/#mz_clusters).
 | `status`       | [`text`]         | The lifecycle status of the reconfiguration: `in-progress` while the controller converges on the target, then a terminal `finalized`, `timed-out`, `cancelled`, or `resource-exhausted`. The record is retained after it settles, so the latest outcome stays inspectable until a later reconfiguration overwrites it. |
 | `deadline`     | [`mz_timestamp`] | The deadline by which the reconfiguration must complete. After it passes, the `on_timeout` action applies. |
 | `on_timeout`   | [`text`]         | The action applied if `deadline` passes before the target hydrates: `commit` (cut over to the not-yet-hydrated target) or `rollback` (revert to the pre-reconfiguration shape). |
-| `target`       | [`jsonb`]        | The config shape the cluster is reconfiguring to, as JSON: `size`, `replication_factor`, `availability_zones`, and `logging`. The realized (current) shape is in `mz_clusters`. |
+| `target`       | [`jsonb`]        | The config shape the cluster is reconfiguring to, as JSON: `size`, `replication_factor`, `availability_zones`, `logging`, and `arrangement_compression`. The realized (current) shape is in `mz_clusters`. |
 | `changes`      | [`jsonb`]        | The dimensions in which `target` differs from the cluster's realized configuration, as a JSON object holding the target value per changed dimension. Empty (`{}`) once a record settles with its target applied. A rolled-back record keeps the abandoned diff. |
 
 ## `mz_cluster_auto_scaling_strategies`

src/adapter/src/catalog/open/builtin_schema_migration.rs

@@ -372,6 +372,16 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
             MZ_CATALOG_SCHEMA,
             "mz_iceberg_sinks",
         ),
+        // The mz_cluster_reconfigurations MV definition changed (the `changes`
+        // diff now includes the `arrangement_compression` dimension). See the
+        // NOTE above: this version must stay at the workspace's current dev
+        // version until the change ships.
+        MigrationStep::replacement(
+            "26.39.0-dev.0",
+            CatalogItemType::MaterializedView,
+            MZ_INTERNAL_SCHEMA,
+            "mz_cluster_reconfigurations",
+        ),
     ]
 });
 

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

@@ -329,7 +329,7 @@ impl Coordinator {
             return Err(AdapterError::AlterClusterScheduleWhileReconfiguring);
         }
 
-        // Replication factor is one of the four dimensions the cut-over sets
+        // Replication factor is one of the dimensions the cut-over sets
         // atomically from the record's target (`fold_reconfiguration_target`),
         // so a change applied independently while a reconfiguration is in
         // flight would be silently clobbered at cut-over. Refused even when the
@@ -797,10 +797,10 @@ impl Coordinator {
         };
 
         // Build the durable write from `new_config`, which carries every field the
-        // `ALTER` changed, then reset the config *shape* (size, replication factor,
-        // availability zones, logging) back to the realized values: that transition
-        // is deferred to the `reconfiguration` record and applied at cut-over. This
-        // applies non-shape changes (`workload_class`, `schedule`,
+        // `ALTER` changed, then reset the config *shape* (every
+        // `ReconfigurationTarget` dimension) back to the realized values: that
+        // transition is deferred to the `reconfiguration` record and applied at
+        // cut-over. This applies non-shape changes (`workload_class`, `schedule`,
         // `auto_scaling_strategy`, ...) immediately, matching the legacy path,
         // rather than silently dropping them. Any existing record is folded over by
         // the `record` we just built.
@@ -811,10 +811,7 @@ impl Coordinator {
                 "reshape_alter_cluster_managed requires a managed realized config".into(),
             ));
         };
-        let realized_size = realized_now.size.clone();
-        let realized_replication_factor = realized_now.replication_factor;
-        let realized_availability_zones = realized_now.availability_zones.clone();
-        let realized_logging = realized_now.logging.clone();
+        let realized_target = realized_now.realized_reconfiguration_target();
         // The status and the audit intent are two views of the same decision,
         // made together here: an ALTER back to the realized shape is a cancel,
         // anything else starts (or re-targets) a reconfiguration.
@@ -842,10 +839,7 @@ impl Coordinator {
                 "reshape_alter_cluster_managed requires a managed target config".into(),
             ));
         };
-        realized_managed.size = realized_size;
-        realized_managed.replication_factor = realized_replication_factor;
-        realized_managed.availability_zones = realized_availability_zones;
-        realized_managed.logging = realized_logging;
+        realized_managed.apply_reconfiguration_target(realized_target);
         realized_managed.reconfiguration = Some(record);
 
         self.catalog_transact(
@@ -2737,8 +2731,9 @@ fn cancel_carried_reconfiguration(config: &mut ClusterConfig) -> Option<Reconfig
 }
 
 /// Whether an `ALTER` statement sets a replica config shape dimension (`SIZE`,
-/// `AVAILABILITY ZONES`, or either `INTROSPECTION` option), the changes that
-/// need a durable `reconfiguration` record and a hydrate-overlap.
+/// `AVAILABILITY ZONES`, either `INTROSPECTION` option, or `EXPERIMENTAL
+/// ARRANGEMENT COMPRESSION`), the changes that need a durable
+/// `reconfiguration` record and a hydrate-overlap.
 ///
 /// A statement-level check, used while a reconfiguration is in flight: an
 /// `ALTER` back to the realized shape sets a shape option without changing its

src/catalog/src/builtin/mz_internal.rs

@@ -876,7 +876,7 @@ pub static MZ_CLUSTER_RECONFIGURATIONS: LazyLock<BuiltinMaterializedView> = Lazy
             ),
             (
                 "target",
-                "The config shape the cluster is reconfiguring to, as JSON: `size`, `replication_factor`, `availability_zones`, and `logging`. The realized (current) shape is in `mz_clusters`.",
+                "The config shape the cluster is reconfiguring to, as JSON: `size`, `replication_factor`, `availability_zones`, `logging`, and `arrangement_compression`. The realized (current) shape is in `mz_clusters`.",
             ),
             (
                 "changes",
@@ -953,7 +953,9 @@ SELECT
     CASE WHEN r.target->'availability_zones' != r.config->'availability_zones'
         THEN jsonb_build_object('availability_zones', r.target->'availability_zones') ELSE '{}'::jsonb END ||
     CASE WHEN r.target->'logging' != r.config->'logging'
-        THEN jsonb_build_object('logging', r.target->'logging') ELSE '{}'::jsonb END
+        THEN jsonb_build_object('logging', r.target->'logging') ELSE '{}'::jsonb END ||
+    CASE WHEN r.target->'arrangement_compression' != r.config->'arrangement_compression'
+        THEN jsonb_build_object('arrangement_compression', r.target->'arrangement_compression') ELSE '{}'::jsonb END
     AS changes
 FROM records r",
         is_retained_metrics_object: false,
@@ -5651,7 +5653,8 @@ pub static MZ_SHOW_CLUSTERS: LazyLock<BuiltinView> = LazyLock::new(|| {
                 'size to ' || (changes->>'size'),
                 'replication factor to ' || (changes->>'replication_factor'),
                 CASE WHEN changes->'availability_zones' IS NOT NULL THEN 'availability zones' END,
-                CASE WHEN changes->'logging' IS NOT NULL THEN 'introspection settings' END
+                CASE WHEN changes->'logging' IS NOT NULL THEN 'introspection settings' END,
+                CASE WHEN changes->'arrangement_compression' IS NOT NULL THEN 'arrangement compression' END
             ], ', '), '') AS summary
         FROM mz_internal.mz_cluster_reconfigurations
         WHERE status = 'in-progress'

src/catalog/src/memory/objects.rs

@@ -3433,6 +3433,26 @@ impl ClusterVariantManaged {
         }
     }
 
+    /// Advances this config's realized shape to `target`, the write a
+    /// reconfiguration cut-over performs. The inverse of
+    /// [`ClusterVariantManaged::realized_reconfiguration_target`].
+    pub fn apply_reconfiguration_target(&mut self, target: ReconfigurationTarget) {
+        // Destructured so a new target dimension fails to compile until it is
+        // applied here too.
+        let ReconfigurationTarget {
+            size,
+            replication_factor,
+            availability_zones,
+            logging,
+            arrangement_compression,
+        } = target;
+        self.size = size;
+        self.replication_factor = replication_factor;
+        self.availability_zones = availability_zones;
+        self.logging = logging;
+        self.arrangement_compression = arrangement_compression;
+    }
+
     /// Whether the in-flight `burst` record is no longer warranted by this
     /// config: the `ON HYDRATION` policy was removed or re-sized away from the
     /// record's size, or the cluster was turned off (`replication_factor` 0).

src/cluster-controller/src/ctx.rs

@@ -219,9 +219,10 @@ pub struct RefreshWindowInputs {
 /// This is the input every strategy reads. Unmanaged clusters are not
 /// controller-owned and are not represented here.
 ///
-/// The `size`, `replication_factor`, `availability_zones`, and `logging` fields
-/// together are the realized config the cluster is currently serving. The
-/// implicit baseline desires `replication_factor` replicas at that shape.
+/// The `size`, `replication_factor`, `availability_zones`, `logging`, and
+/// `arrangement_compression` fields together are the realized config the
+/// cluster is currently serving. The implicit baseline desires
+/// `replication_factor` replicas at that shape.
 #[derive(Clone, Debug)]
 pub struct ClusterState {
     pub cluster_id: ClusterId,

src/sql/src/plan/statement/ddl.rs

@@ -6550,10 +6550,11 @@ pub fn plan_alter_cluster(
                         && availability_zones.is_none()
                         && introspection_debugging.is_none()
                         && introspection_interval.is_none()
+                        && experimental_arrangement_compression.is_none()
                     {
                         sql_bail!(
                             "WAIT can only be used together with a SIZE, AVAILABILITY ZONES, \
-                            or INTROSPECTION change"
+                            INTROSPECTION, or EXPERIMENTAL ARRANGEMENT COMPRESSION change"
                         );
                     }
 

test/sqllogictest/autogenerated/mz_internal.slt

@@ -129,7 +129,7 @@ cluster_id  text  The␠ID␠of␠the␠cluster.␠Corresponds␠to␠`mz_cluste
 status  text  The␠lifecycle␠status␠of␠the␠reconfiguration:␠`in-progress`␠while␠the␠controller␠converges␠on␠the␠target,␠then␠a␠terminal␠`finalized`,␠`timed-out`,␠`cancelled`,␠or␠`resource-exhausted`.␠The␠record␠is␠retained␠after␠it␠settles,␠so␠the␠latest␠outcome␠stays␠inspectable␠until␠a␠later␠reconfiguration␠overwrites␠it.
 deadline  mz_timestamp  The␠deadline␠by␠which␠the␠reconfiguration␠must␠complete.␠After␠it␠passes,␠the␠`on_timeout`␠action␠applies.
 on_timeout  text  The␠action␠applied␠if␠`deadline`␠passes␠before␠the␠target␠hydrates:␠`commit`␠(cut␠over␠to␠the␠not-yet-hydrated␠target)␠or␠`rollback`␠(revert␠to␠the␠pre-reconfiguration␠shape).
-target  jsonb  The␠config␠shape␠the␠cluster␠is␠reconfiguring␠to,␠as␠JSON:␠`size`,␠`replication_factor`,␠`availability_zones`,␠and␠`logging`.␠The␠realized␠(current)␠shape␠is␠in␠`mz_clusters`.
+target  jsonb  The␠config␠shape␠the␠cluster␠is␠reconfiguring␠to,␠as␠JSON:␠`size`,␠`replication_factor`,␠`availability_zones`,␠`logging`,␠and␠`arrangement_compression`.␠The␠realized␠(current)␠shape␠is␠in␠`mz_clusters`.
 changes  jsonb  The␠dimensions␠in␠which␠`target`␠differs␠from␠the␠cluster's␠realized␠configuration,␠as␠a␠JSON␠object␠holding␠the␠target␠value␠per␠changed␠dimension.␠Empty␠(`{}`)␠once␠a␠record␠settles␠with␠its␠target␠applied.␠A␠rolled-back␠record␠keeps␠the␠abandoned␠diff.
 
 query TTT

test/sqllogictest/catalog_server_explain.slt

@@ -2009,7 +2009,7 @@ mz_internal.mz_show_clusters:
           →Fused with Child Map/Filter/Project
             Project: #0, #6, #7
             Filter: (#1{status} = "in-progress")
-            Map: ("reconfiguring " || case when ("" = array_to_string(array[("size to " || (#5{changes} ->> "size")), ("replication factor to " || (#5{changes} ->> "replication_factor")), case when ((#5{changes} -> "availability_zones")) IS NOT NULL then "availability zones" else null end, case when ((#5{changes} -> "logging")) IS NOT NULL then "introspection settings" else null end], ", ")) then null else array_to_string(array[("size to " || (#5{changes} ->> "size")), ("replication factor to " || (#5{changes} ->> "replication_factor")), case when ((#5{changes} -> "availability_zones")) IS NOT NULL then "availability zones" else null end, case when ((#5{changes} -> "logging")) IS NOT NULL then "introspection settings" else null end], ", ") end), true
+            Map: ("reconfiguring " || case when ("" = array_to_string(array[("size to " || (#5{changes} ->> "size")), ("replication factor to " || (#5{changes} ->> "replication_factor")), case when ((#5{changes} -> "availability_zones")) IS NOT NULL then "availability zones" else null end, case when ((#5{changes} -> "logging")) IS NOT NULL then "introspection settings" else null end, case when ((#5{changes} -> "arrangement_compression")) IS NOT NULL then "arrangement compression" else null end], ", ")) then null else array_to_string(array[("size to " || (#5{changes} ->> "size")), ("replication factor to " || (#5{changes} ->> "replication_factor")), case when ((#5{changes} -> "availability_zones")) IS NOT NULL then "availability zones" else null end, case when ((#5{changes} -> "logging")) IS NOT NULL then "introspection settings" else null end, case when ((#5{changes} -> "arrangement_compression")) IS NOT NULL then "arrangement compression" else null end], ", ") end), true
               →Arranged mz_internal.mz_cluster_reconfigurations
                 Key: (#0{cluster_id})
           →Map/Filter/Project
@@ -5371,7 +5371,7 @@ mz_internal.mz_cluster_reconfigurations:
 Source mz_internal.mz_catalog_raw
   project=(#3, #5..=#7, #4, #8)
   filter=(("Cluster" = (#0{data} ->> "kind")) AND (json_null != #2) AND (#2 != json_null))
-  map=(((((#0{data} -> "value") -> "config") -> "variant") -> "Managed"), (#1 -> "reconfiguration"), parse_catalog_id(((#0{data} -> "key") -> "id")), (#2 -> "target"), case when ("InProgress" = (#2{reconfiguration} ->> "status")) then "in-progress" else case when ("Finalized" = (#2 ->> "status")) then "finalized" else case when ("TimedOut" = (#2 ->> "status")) then "timed-out" else case when ("Cancelled" = (#2 ->> "status")) then "cancelled" else case when ("ResourceExhausted" = (#2 ->> "status")) then "resource-exhausted" else (#2 ->> "status") end end end end end, text_to_mz_timestamp((#2{reconfiguration} ->> "deadline")), case when ("Commit" = (#2{reconfiguration} ->> "on_timeout")) then "commit" else case when ("Rollback" = (#2 ->> "on_timeout")) then "rollback" else (#2 ->> "on_timeout") end end, (((case when ((#1{config} -> "size") != (#4{target} -> "size")) then jsonb_build_object("size", jsonbable_to_jsonb((#4 -> "size"))) else {} end || case when ((#1{config} -> "replication_factor") != (#4{target} -> "replication_factor")) then jsonb_build_object("replication_factor", jsonbable_to_jsonb((#4 -> "replication_factor"))) else {} end) || case when ((#1{config} -> "availability_zones") != (#4{target} -> "availability_zones")) then jsonb_build_object("availability_zones", jsonbable_to_jsonb((#4 -> "availability_zones"))) else {} end) || case when ((#1{config} -> "logging") != (#4{target} -> "logging")) then jsonb_build_object("logging", jsonbable_to_jsonb((#4 -> "logging"))) else {} end))
+  map=(((((#0{data} -> "value") -> "config") -> "variant") -> "Managed"), (#1 -> "reconfiguration"), parse_catalog_id(((#0{data} -> "key") -> "id")), (#2 -> "target"), case when ("InProgress" = (#2{reconfiguration} ->> "status")) then "in-progress" else case when ("Finalized" = (#2 ->> "status")) then "finalized" else case when ("TimedOut" = (#2 ->> "status")) then "timed-out" else case when ("Cancelled" = (#2 ->> "status")) then "cancelled" else case when ("ResourceExhausted" = (#2 ->> "status")) then "resource-exhausted" else (#2 ->> "status") end end end end end, text_to_mz_timestamp((#2{reconfiguration} ->> "deadline")), case when ("Commit" = (#2{reconfiguration} ->> "on_timeout")) then "commit" else case when ("Rollback" = (#2 ->> "on_timeout")) then "rollback" else (#2 ->> "on_timeout") end end, ((((case when ((#1{config} -> "size") != (#4{target} -> "size")) then jsonb_build_object("size", jsonbable_to_jsonb((#4 -> "size"))) else {} end || case when ((#1{config} -> "replication_factor") != (#4{target} -> "replication_factor")) then jsonb_build_object("replication_factor", jsonbable_to_jsonb((#4 -> "replication_factor"))) else {} end) || case when ((#1{config} -> "availability_zones") != (#4{target} -> "availability_zones")) then jsonb_build_object("availability_zones", jsonbable_to_jsonb((#4 -> "availability_zones"))) else {} end) || case when ((#1{config} -> "logging") != (#4{target} -> "logging")) then jsonb_build_object("logging", jsonbable_to_jsonb((#4 -> "logging"))) else {} end) || case when ((#1{config} -> "arrangement_compression") != (#4{target} -> "arrangement_compression")) then jsonb_build_object("arrangement_compression", jsonbable_to_jsonb((#4 -> "arrangement_compression"))) else {} end))
 
 Target cluster: mz_catalog_server