adapter,test: support forcing builtin schema migrations

This commit adds support for forcing builtin schema migrations during
tests, through a --force-builtin-schema-migration envd flag, which gets
wired through to the builtin schema migration code. If the flag is set,
all builtin collections are force-migrated using the selected mechanism.

The existing mechanism to force builtin item migrations worked by
modifying the fingerprints of builtins. This doesn't work anymore
because fingerprint mismatches are no longer used to inform which
migrations to run, only to verify all items have been correctly
migrated. Thus, this commit also removes support for adding whitespace
to builtin fingerprints.

Author: teskje

misc/python/materialize/checks/scenarios_zero_downtime.py

@@ -93,7 +93,7 @@ def actions(self) -> list[Action]:
                 deploy_generation=1,
                 mz_service="mz_2",
                 system_parameter_defaults=system_parameter_defaults,
-                force_migrations="all",
+                force_migrations="replacement",
             ),
             Manipulate(self, phase=1, mz_service="mz_1"),
             *wait_ready_and_promote("mz_2"),
@@ -102,7 +102,7 @@ def actions(self) -> list[Action]:
                 deploy_generation=2,
                 mz_service="mz_3",
                 system_parameter_defaults=system_parameter_defaults,
-                force_migrations="all",
+                force_migrations="replacement",
             ),
             Manipulate(self, phase=2, mz_service="mz_2"),
             *wait_ready_and_promote("mz_3"),
@@ -111,7 +111,7 @@ def actions(self) -> list[Action]:
                 deploy_generation=3,
                 mz_service="mz_4",
                 system_parameter_defaults=system_parameter_defaults,
-                force_migrations="all",
+                force_migrations="replacement",
             ),
             Validate(self, mz_service="mz_3"),
             *wait_ready_and_promote("mz_4"),

misc/python/materialize/mzcompose/services/materialized.py

@@ -220,7 +220,7 @@ def __init__(
 
         if force_migrations is not None and image is None:
             command += [
-                f"--unsafe-builtin-table-fingerprint-whitespace={force_migrations}",
+                f"--unsafe-force-builtin-schema-migration={force_migrations}",
             ]
             if not unsafe_mode:
                 command += ["--unsafe-mode"]

src/adapter/src/catalog.rs

@@ -721,6 +721,7 @@ impl Catalog {
                 builtin_item_migration_config: BuiltinItemMigrationConfig {
                     persist_client: persist_client.clone(),
                     read_only,
+                    force_migration: None,
                 },
                 persist_client,
                 enable_expression_cache_override,

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

@@ -240,14 +240,20 @@ impl<'a, 'b> Migration<'a, 'b> {
 
         self.validate_migration_steps(steps);
 
-        if self.source_version == self.target_version {
+        let (force, plan) = match self.config.force_migration.as_deref() {
+            None => (false, self.plan_migration(steps)),
+            Some("evolution") => (true, self.plan_forced_migration(Mechanism::Evolution)),
+            Some("replacement") => (true, self.plan_forced_migration(Mechanism::Replacement)),
+            Some(other) => panic!("unknown force migration mechanism: {other}"),
+        };
+
+        if self.source_version == self.target_version && !force {
             info!("skipping migration: already at target version");
             return Ok(Default::default());
         } else if self.source_version > self.target_version {
             bail!("downgrade not supported");
         }
 
-        let plan = self.plan_migration(steps);
         info!("executing migration plan: {plan:?}");
 
         self.migrate_evolve(&plan.evolve).await?;
@@ -346,6 +352,24 @@ impl<'a, 'b> Migration<'a, 'b> {
         plan
     }
 
+    /// Plan a forced migration of all objects using the given mechanism.
+    fn plan_forced_migration(&self, mechanism: Mechanism) -> Plan {
+        let objects = self
+            .builtins
+            .iter()
+            .filter(|(_, builtin)| matches!(builtin, Builtin::Table(..) | Builtin::Source(..)))
+            .map(|(object, _)| object.clone())
+            .collect();
+
+        let mut plan = Plan::default();
+        match mechanism {
+            Mechanism::Evolution => plan.evolve = objects,
+            Mechanism::Replacement => plan.replace = objects,
+        }
+
+        plan
+    }
+
     /// Migrate the given objects using the `Evolution` mechanism.
     async fn migrate_evolve(&self, objects: &[SystemObjectDescription]) -> anyhow::Result<()> {
         for object in objects {

src/adapter/src/coord.rs

@@ -1066,6 +1066,7 @@ pub struct Config {
     pub helm_chart_version: Option<String>,
     pub license_key: ValidatedLicenseKey,
     pub external_login_password_mz_system: Option<Password>,
+    pub force_builtin_schema_migration: Option<String>,
 }
 
 /// Metadata about an active connection.
@@ -4014,6 +4015,7 @@ pub fn serve(
         helm_chart_version,
         license_key,
         external_login_password_mz_system,
+        force_builtin_schema_migration,
     }: Config,
 ) -> BoxFuture<'static, Result<(Handle, Client), AdapterError>> {
     async move {
@@ -4121,6 +4123,7 @@ pub fn serve(
             BuiltinItemMigrationConfig {
                 persist_client: persist_client.clone(),
                 read_only: read_only_controllers,
+                force_migration: force_builtin_schema_migration,
             }
         ;
         let OpenCatalogResult {

src/catalog-debug/src/main.rs

@@ -651,6 +651,7 @@ async fn upgrade_check(
                 // client.
                 persist_client: PersistClient::new_for_tests().await,
                 read_only,
+                force_migration: None,
             },
             persist_client: persist_client.clone(),
             enable_expression_cache_override: None,

src/catalog/src/builtin.rs

@@ -28,9 +28,7 @@ use std::collections::BTreeMap;
 use std::hash::Hash;
 use std::string::ToString;
 use std::sync::LazyLock;
-use std::sync::Mutex;
 
-use clap::clap_derive::ValueEnum;
 use mz_compute_client::logging::{ComputeLog, DifferentialLog, LogVariant, TimelyLog};
 use mz_ore::collections::HashMap;
 use mz_pgrepr::oid;
@@ -61,7 +59,6 @@ use mz_storage_client::healthcheck::{
     REPLICA_STATUS_HISTORY_DESC, WALLCLOCK_LAG_HISTORY_DESC,
 };
 use mz_storage_client::statistics::{MZ_SINK_STATISTICS_RAW_DESC, MZ_SOURCE_STATISTICS_RAW_DESC};
-use rand::Rng;
 use serde::Serialize;
 
 use crate::durable::objects::SystemObjectDescription;
@@ -350,48 +347,9 @@ impl Fingerprint for &BuiltinLog {
     }
 }
 
-/// Allows tests to inject arbitrary amounts of whitespace to forcibly change the fingerprint and
-/// trigger a builtin migration.
-#[derive(Debug, Clone, ValueEnum)]
-pub enum UnsafeBuiltinTableFingerprintWhitespace {
-    /// Inject whitespace into all builtin table fingerprints.
-    All,
-    /// Inject whitespace into half of the builtin table fingerprints,
-    /// which are randomly selected.
-    Half,
-}
-pub static UNSAFE_DO_NOT_CALL_THIS_IN_PRODUCTION_BUILTIN_TABLE_FINGERPRINT_WHITESPACE: Mutex<
-    Option<(UnsafeBuiltinTableFingerprintWhitespace, String)>,
-> = Mutex::new(None);
-
 impl Fingerprint for &BuiltinTable {
     fn fingerprint(&self) -> String {
-        // This is only called during bootstrapping, so it's not that big of a deal to lock a mutex,
-        // though it's not great.
-        let guard = UNSAFE_DO_NOT_CALL_THIS_IN_PRODUCTION_BUILTIN_TABLE_FINGERPRINT_WHITESPACE
-            .lock()
-            .expect("lock poisoned");
-        match &*guard {
-            // `mz_storage_usage_by_shard` can never be migrated.
-            _ if self.schema == MZ_STORAGE_USAGE_BY_SHARD.schema
-                && self.name == MZ_STORAGE_USAGE_BY_SHARD.name =>
-            {
-                self.desc.fingerprint()
-            }
-            Some((UnsafeBuiltinTableFingerprintWhitespace::All, whitespace)) => {
-                format!("{}{}", self.desc.fingerprint(), whitespace)
-            }
-            Some((UnsafeBuiltinTableFingerprintWhitespace::Half, whitespace)) => {
-                let mut rng = rand::thread_rng();
-                let migrate: bool = rng.r#gen();
-                if migrate {
-                    format!("{}{}", self.desc.fingerprint(), whitespace)
-                } else {
-                    self.desc.fingerprint()
-                }
-            }
-            None => self.desc.fingerprint(),
-        }
+        self.desc.fingerprint()
     }
 }
 

src/catalog/src/config.rs

@@ -104,6 +104,7 @@ pub struct StateConfig {
 pub struct BuiltinItemMigrationConfig {
     pub persist_client: PersistClient,
     pub read_only: bool,
+    pub force_migration: Option<String>,
 }
 
 #[derive(Debug, Clone, Serialize)]

src/environmentd/src/environmentd/main.rs

@@ -38,10 +38,6 @@ use mz_adapter_types::bootstrap_builtin_cluster_config::{
 use mz_auth::password::Password;
 use mz_aws_secrets_controller::AwsSecretsController;
 use mz_build_info::BuildInfo;
-use mz_catalog::builtin::{
-    UNSAFE_DO_NOT_CALL_THIS_IN_PRODUCTION_BUILTIN_TABLE_FINGERPRINT_WHITESPACE,
-    UnsafeBuiltinTableFingerprintWhitespace,
-};
 use mz_catalog::config::ClusterReplicaSizeMap;
 use mz_cloud_resources::{AwsExternalIdPrefix, CloudResourceController};
 use mz_controller::ControllerConfig;
@@ -617,21 +613,13 @@ pub struct Args {
     tracing: TracingCliArgs,
 
     // === Testing options. ===
-    /// Injects arbitrary whitespace into builtin table fingerprints, which can
-    /// trigger builtin item migrations. The amount of whitespace is determined
-    /// by
-    /// `unsafe_builtin_table_fingerprint_whitespace_version`.
+    /// Forces the migration of all builtin storage collections using the
+    /// specified migration mechanism (either "evolution" or "replacement").
+    ///
     /// This argument is meant for testing only and as the names suggests
     /// should not be set in production.
     #[clap(long, value_enum, requires = "unsafe_mode")]
-    unsafe_builtin_table_fingerprint_whitespace: Option<UnsafeBuiltinTableFingerprintWhitespace>,
-    /// Controls the amount of whitespace injected by
-    /// `unsafe_builtin_table_fingerprint_whitespace`.
-    /// Incrementing this value can allow triggering multiple builtin
-    /// migrations from a single test. This argument is meant for testing only
-    /// and as the names suggests should not be set in production.
-    #[clap(long, requires = "unsafe_mode", default_value = "1")]
-    unsafe_builtin_table_fingerprint_whitespace_version: usize,
+    unsafe_force_builtin_schema_migration: Option<String>,
 }
 
 #[derive(ValueEnum, Debug, Clone)]
@@ -695,13 +683,9 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> {
     };
 
     // Configure testing options.
-    if let Some(fingerprint_whitespace) = args.unsafe_builtin_table_fingerprint_whitespace {
-        assert!(args.unsafe_mode);
-        let whitespace = "\n".repeat(args.unsafe_builtin_table_fingerprint_whitespace_version);
-        *UNSAFE_DO_NOT_CALL_THIS_IN_PRODUCTION_BUILTIN_TABLE_FINGERPRINT_WHITESPACE
-            .lock()
-            .expect("lock poisoned") = Some((fingerprint_whitespace, whitespace));
-    }
+    let force_builtin_schema_migration = args
+        .unsafe_force_builtin_schema_migration
+        .inspect(|_mechanism| assert!(args.unsafe_mode));
 
     // Start Tokio runtime.
 
@@ -1168,6 +1152,7 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> {
                 tracing_handle,
                 // Testing options.
                 now,
+                force_builtin_schema_migration,
             })
             .await
             .maybe_terminate("booting server")?;

src/environmentd/src/lib.rs

@@ -209,6 +209,9 @@ pub struct Config {
     // === Testing options. ===
     /// A now generation function for mocking time.
     pub now: NowFn,
+    /// If `Some`, force running builtin schema migration using the specified
+    /// migration mechanism ("evolution" or "replacement").
+    pub force_builtin_schema_migration: Option<String>,
 }
 
 /// Configuration for the Catalog.
@@ -786,6 +789,7 @@ impl Listeners {
             helm_chart_version: config.helm_chart_version.clone(),
             license_key: config.license_key,
             external_login_password_mz_system: config.external_login_password_mz_system,
+            force_builtin_schema_migration: config.force_builtin_schema_migration,
         })
         .instrument(info_span!("adapter::serve"))
         .await?;