mz_catalog/durable/

upgrade.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

//! This module contains all the helpers and code paths for upgrading/migrating the `Catalog`.
//!
//! We facilitate migrations by keeping snapshots of the objects we previously stored, and relying
//! entirely on these snapshots. These snapshots exist in the [`mz_catalog_protos`] crate in the
//! form of `catalog-protos/protos/objects_vXX.proto`. By maintaining and relying on snapshots we
//! don't have to worry about changes elsewhere in the codebase effecting our migrations because
//! our application and serialization logic is decoupled, and the objects of the Catalog for a
//! given version are "frozen in time".
//!
//! > **Note**: The protobuf snapshot files themselves live in a separate crate because it takes a
//!             relatively significant amount of time to codegen and build them. By placing them in
//!             a separate crate we don't have to pay this compile time cost when building the
//!             catalog, allowing for faster iteration.
//!
//! You cannot make any changes to the following message or anything that they depend on:
//!
//!   - Config
//!   - Setting
//!   - FenceToken
//!   - AuditLog
//!
//! When you want to make a change to the `Catalog` you need to follow these steps:
//!
//! 1. Check the current [`CATALOG_VERSION`], make sure an `objects_v<CATALOG_VERSION>.proto` file
//!    exists. If one doesn't, copy and paste the current `objects.proto` file, renaming it to
//!    `objects_v<CATALOG_VERSION>.proto`.
//! 2. Bump [`CATALOG_VERSION`] by one.
//! 3. Make your changes to `objects.proto`.
//! 4. Copy and paste `objects.proto`, naming the copy `objects_v<CATALOG_VERSION>.proto`. Update
//!    the package name of the `.proto` to `package objects_v<CATALOG_VERSION>;`
//! 5. We should now have a copy of the protobuf objects as they currently exist, and a copy of
//!    how we want them to exist. For example, if the version of the Catalog before we made our
//!    changes was 15, we should now have `objects_v15.proto` and `objects_v16.proto`.
//! 6. Rebuild Materialize which will error because the hashes stored in
//!    `src/catalog-protos/protos/hashes.json` have now changed. Update these to match the new
//!    hashes for objects.proto and `objects_v<CATALOG_VERSION>.proto`.
//! 7. Add `v<CATALOG_VERSION>` to the call to the `objects!` macro in this file
//! 8. Add a new file to `catalog/src/durable/upgrade` which is where we'll put the new migration
//!    path.
//! 9. Write upgrade functions using the two versions of the protos we now have, e.g.
//!    `objects_v15.proto` and `objects_v16.proto`. In this migration code you __should not__
//!    import any defaults or constants from elsewhere in the codebase, because then a future
//!    change could then impact a previous migration.
//! 10. Add an import for your new module to this file: mod v<CATALOG_VERSION-1>_to_v<CATALOG_VERSION>;
//! 11. Call your upgrade function in [`run_upgrade()`].
//! 12. Generate a test file for the new version:
//!     ```ignore
//!     cargo test --package mz-catalog --lib durable::upgrade::tests::generate_missing_encodings -- --ignored
//!     ```
//!
//! When in doubt, reach out to the Surfaces team, and we'll be more than happy to help :)

pub mod json_compatible;
#[cfg(test)]
mod tests;

use mz_ore::{soft_assert_eq_or_log, soft_assert_ne_or_log};
use mz_repr::Diff;
use paste::paste;
#[cfg(test)]
use proptest::prelude::*;
#[cfg(test)]
use proptest::strategy::ValueTree;
#[cfg(test)]
use proptest_derive::Arbitrary;
use timely::progress::Timestamp as TimelyTimestamp;

use crate::durable::initialize::USER_VERSION_KEY;
use crate::durable::objects::serialization::proto;
use crate::durable::objects::state_update::{
    IntoStateUpdateKindJson, StateUpdate, StateUpdateKind, StateUpdateKindJson,
};
use crate::durable::persist::{Mode, Timestamp, UnopenedPersistCatalogState};
use crate::durable::{CatalogError, DurableCatalogError};

#[cfg(test)]
const ENCODED_TEST_CASES: usize = 100;

/// Generate per-version support code.
///
/// Here we have to deal with the fact that the pre-v79 objects had a protobuf-generated format,
/// which gives them additional levels of indirection that the post-v79 objects don't have and thus
/// requires slightly different code to be generated.
macro_rules! objects {
    ( [$( $x_old:ident ),*], [$( $x:ident ),*] ) => {
        paste! {
            $(
                pub(crate) mod [<objects_ $x_old>] {
                    pub use mz_catalog_protos::[<objects_ $x_old>]::*;

                    use crate::durable::objects::state_update::StateUpdateKindJson;

                    impl From<StateUpdateKind> for StateUpdateKindJson {
                        fn from(value: StateUpdateKind) -> Self {
                            let kind = value.kind.expect("kind should be set");
                            // TODO: This requires that the json->proto->json roundtrips
                            // exactly, see SQL-895.
                            StateUpdateKindJson::from_serde(&kind)
                        }
                    }

                    impl From<StateUpdateKindJson> for StateUpdateKind {
                        fn from(value: StateUpdateKindJson) -> Self {
                            let kind: state_update_kind::Kind = value.to_serde();
                            StateUpdateKind { kind: Some(kind) }
                        }
                    }
                }
            )*

            $(
                pub(crate) mod [<objects_ $x>] {
                    pub use mz_catalog_protos::[<objects_ $x>]::*;

                    use crate::durable::objects::state_update::StateUpdateKindJson;

                    impl From<StateUpdateKind> for StateUpdateKindJson {
                        fn from(value: StateUpdateKind) -> Self {
                            // TODO: This requires that the json->proto->json roundtrips
                            // exactly, see SQL-895.
                            StateUpdateKindJson::from_serde(&value)
                        }
                    }

                    impl From<StateUpdateKindJson> for StateUpdateKind {
                        fn from(value: StateUpdateKindJson) -> Self {
                            value.to_serde()
                        }
                    }
                }
            )*

            // Generate test helpers for each version.

            #[cfg(test)]
            #[derive(Debug, Arbitrary)]
            enum AllVersionsStateUpdateKind {
                $(
                    [<$x_old:upper>](crate::durable::upgrade::[<objects_ $x_old>]::StateUpdateKind),
                )*
                $(
                    [<$x:upper>](crate::durable::upgrade::[<objects_ $x>]::StateUpdateKind),
                )*
            }

            #[cfg(test)]
            impl AllVersionsStateUpdateKind {
                #[cfg(test)]
                fn arbitrary_vec(version: &str) -> Result<Vec<Self>, String> {
                    let mut runner = proptest::test_runner::TestRunner::deterministic();
                    std::iter::repeat(())
                        .filter_map(|_| {
                            AllVersionsStateUpdateKind::arbitrary(version, &mut runner)
                                .transpose()
                        })
                        .take(ENCODED_TEST_CASES)
                        .collect::<Result<_, _>>()
                }

                #[cfg(test)]
                fn arbitrary(
                    version: &str,
                    runner: &mut proptest::test_runner::TestRunner,
                ) -> Result<Option<Self>, String> {
                    match version {
                        $(
                            concat!("objects_", stringify!($x_old)) => {
                                let arbitrary_data =
                                    crate::durable::upgrade
                                        ::[<objects_ $x_old>]::StateUpdateKind::arbitrary()
                                        .new_tree(runner)
                                        .expect("unable to create arbitrary data")
                                        .current();
                                // Skip over generated data where kind is None
                                // because they are not interesting or possible in
                                // production. Unfortunately any of the inner fields
                                // can still be None, which is also not possible in
                                // production.
                                // TODO(jkosh44) See if there's an arbitrary config
                                // that forces Some.
                                let arbitrary_data = if arbitrary_data.kind.is_some() {
                                    Some(Self::[<$x_old:upper>](arbitrary_data))
                                } else {
                                    None
                                };
                                Ok(arbitrary_data)
                            }
                        )*
                        $(
                            concat!("objects_", stringify!($x)) => {
                                let arbitrary_data =
                                    crate::durable::upgrade
                                        ::[<objects_ $x>]::StateUpdateKind::arbitrary()
                                        .new_tree(runner)
                                        .expect("unable to create arbitrary data")
                                        .current();
                                Ok(Some(Self::[<$x:upper>](arbitrary_data)))
                            }
                        )*
                        _ => Err(format!("unrecognized version {version} add enum variant")),
                    }
                }

                #[cfg(test)]
                fn try_from_raw(version: &str, raw: StateUpdateKindJson) -> Result<Self, String> {
                    match version {
                        $(
                            concat!("objects_", stringify!($x_old)) => Ok(Self::[<$x_old:upper>](raw.into())),
                        )*
                        $(
                            concat!("objects_", stringify!($x)) => Ok(Self::[<$x:upper>](raw.into())),

                        )*
                        _ => Err(format!("unrecognized version {version} add enum variant")),
                    }
                }

                #[cfg(test)]
                fn raw(self) -> StateUpdateKindJson {
                    match self {
                        $(
                            Self::[<$x_old:upper>](kind) => kind.into(),
                        )*
                        $(
                            Self::[<$x:upper>](kind) => kind.into(),
                        )*
                    }
                }
            }
        }
    }
}

objects!(
    [v74, v75, v76, v77, v78],
    [
        v79, v80, v81, v82, v83, v84, v85, v86, v87, v88, v89, v90, v91, v92, v93, v94, v95
    ]
);

/// The current version of the `Catalog`.
pub use mz_catalog_protos::CATALOG_VERSION;
/// The minimum `Catalog` version number that we support migrating from.
pub use mz_catalog_protos::MIN_CATALOG_VERSION;

// Note(parkmycar): Ideally we wouldn't have to define these extra constants,
// but const expressions aren't yet supported in match statements.
const TOO_OLD_VERSION: u64 = MIN_CATALOG_VERSION - 1;
const FUTURE_VERSION: u64 = CATALOG_VERSION + 1;

mod v74_to_v75;
mod v75_to_v76;
mod v76_to_v77;
mod v77_to_v78;
mod v78_to_v79;
mod v79_to_v80;
mod v80_to_v81;
mod v81_to_v82;
mod v82_to_v83;
mod v83_to_v84;
mod v84_to_v85;
mod v85_to_v86;
mod v86_to_v87;
mod v87_to_v88;
mod v88_to_v89;
mod v89_to_v90;
mod v90_to_v91;
mod v91_to_v92;
mod v92_to_v93;
mod v93_to_v94;
mod v94_to_v95;

/// Describes a single action to take during a migration from `V1` to `V2`.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
enum MigrationAction<V1: IntoStateUpdateKindJson, V2: IntoStateUpdateKindJson> {
    /// Deletes the provided key.
    #[allow(unused)]
    Delete(V1),
    /// Inserts the provided key-value pair. The key must not currently exist!
    #[allow(unused)]
    Insert(V2),
    /// Update the key-value pair for the provided key.
    #[allow(unused)]
    Update(V1, V2),
}

impl<V1: IntoStateUpdateKindJson, V2: IntoStateUpdateKindJson> MigrationAction<V1, V2> {
    /// Converts `self` into a `Vec<StateUpdate<StateUpdateKindBinary>>` that can be appended
    /// to persist.
    fn into_updates(self) -> Vec<(StateUpdateKindJson, Diff)> {
        match self {
            MigrationAction::Delete(kind) => {
                vec![(kind.into(), Diff::MINUS_ONE)]
            }
            MigrationAction::Insert(kind) => {
                vec![(kind.into(), Diff::ONE)]
            }
            MigrationAction::Update(old_kind, new_kind) => {
                vec![
                    (old_kind.into(), Diff::MINUS_ONE),
                    (new_kind.into(), Diff::ONE),
                ]
            }
        }
    }
}

/// Upgrades the data in the catalog to version [`CATALOG_VERSION`].
///
/// Returns the current upper after all migrations have executed.
#[mz_ore::instrument(name = "persist::upgrade", level = "debug")]
pub(crate) async fn upgrade(
    persist_handle: &mut UnopenedPersistCatalogState,
    mut commit_ts: Timestamp,
    force_migration: bool,
) -> Result<Timestamp, CatalogError> {
    soft_assert_ne_or_log!(
        persist_handle.upper,
        Timestamp::minimum(),
        "cannot upgrade uninitialized catalog"
    );

    // Consolidate to avoid migrating old state.
    persist_handle.consolidate();
    let mut version = persist_handle
        .get_user_version()
        .await?
        .expect("initialized catalog must have a version");
    // Run migrations until we're up-to-date.
    while version < CATALOG_VERSION {
        (version, commit_ts) = run_upgrade(persist_handle, version, commit_ts).await?;
    }
    if force_migration {
        tracing::info!(version, "running forced Catalog upgrade");
        commit_ts = apply_migration(persist_handle, Vec::new(), commit_ts).await?;
    }

    Ok(commit_ts)
}

/// Determines which upgrade to run for the `version` and executes it.
///
/// Returns the new version and upper.
async fn run_upgrade(
    unopened_catalog_state: &mut UnopenedPersistCatalogState,
    version: u64,
    commit_ts: Timestamp,
) -> Result<(u64, Timestamp), CatalogError> {
    let incompatible = DurableCatalogError::IncompatibleDataVersion {
        found_version: version,
        min_catalog_version: MIN_CATALOG_VERSION,
        catalog_version: CATALOG_VERSION,
    }
    .into();

    match version {
        ..=TOO_OLD_VERSION => Err(incompatible),

        74 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v74_to_v75::upgrade,
            )
            .await
        }
        75 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v75_to_v76::upgrade,
            )
            .await
        }
        76 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v76_to_v77::upgrade,
            )
            .await
        }
        77 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v77_to_v78::upgrade,
            )
            .await
        }
        78 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v78_to_v79::upgrade,
            )
            .await
        }
        79 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v79_to_v80::upgrade,
            )
            .await
        }
        80 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v80_to_v81::upgrade,
            )
            .await
        }
        81 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v81_to_v82::upgrade,
            )
            .await
        }
        // v82→v83 is a one-shot byte-level repair, not a proto evolution. It
        // needs raw access to the snapshot's diffs (which `run_versioned_upgrade`
        // strips) so it plugs into `run_upgrade` directly.
        82 => v82_to_v83::upgrade(unopened_catalog_state, commit_ts).await,
        83 => v83_to_v84::upgrade(unopened_catalog_state, commit_ts).await,
        84 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v84_to_v85::upgrade,
            )
            .await
        }
        85 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v85_to_v86::upgrade,
            )
            .await
        }
        86 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v86_to_v87::upgrade,
            )
            .await
        }
        87 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v87_to_v88::upgrade,
            )
            .await
        }
        88 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v88_to_v89::upgrade,
            )
            .await
        }
        89 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v89_to_v90::upgrade,
            )
            .await
        }
        90 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v90_to_v91::upgrade,
            )
            .await
        }
        91 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v91_to_v92::upgrade,
            )
            .await
        }
        92 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v92_to_v93::upgrade,
            )
            .await
        }
        93 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v93_to_v94::upgrade,
            )
            .await
        }
        94 => {
            run_versioned_upgrade(
                unopened_catalog_state,
                version,
                commit_ts,
                v94_to_v95::upgrade,
            )
            .await
        }
        // Up-to-date, no migration needed!
        CATALOG_VERSION => Ok((CATALOG_VERSION, commit_ts)),
        FUTURE_VERSION.. => Err(incompatible),
    }
}

/// Runs `migration_logic` on the contents of the current catalog assuming a current version of
/// `current_version`.
///
/// Returns the new version and upper.
async fn run_versioned_upgrade<V1: IntoStateUpdateKindJson, V2: IntoStateUpdateKindJson>(
    unopened_catalog_state: &mut UnopenedPersistCatalogState,
    current_version: u64,
    commit_ts: Timestamp,
    migration_logic: impl FnOnce(Vec<V1>) -> Vec<MigrationAction<V1, V2>>,
) -> Result<(u64, Timestamp), CatalogError> {
    tracing::info!(current_version, "running versioned Catalog upgrade");

    // 1. Use the V1 to deserialize the contents of the current snapshot.
    let snapshot: Vec<_> = unopened_catalog_state
        .snapshot
        .iter()
        .map(|(kind, ts, diff)| {
            soft_assert_eq_or_log!(
                *diff,
                Diff::ONE,
                "snapshot is consolidated, ({kind:?}, {ts:?}, {diff:?})"
            );
            V1::try_from(kind.clone()).expect("invalid catalog data persisted")
        })
        .collect();

    // 2. Generate updates from version specific migration logic.
    let migration_actions = migration_logic(snapshot);
    let mut updates: Vec<_> = migration_actions
        .into_iter()
        .flat_map(|action| action.into_updates().into_iter())
        .collect();
    // Validate that we're not migrating an un-migratable collection.
    for (update, _) in &updates {
        if update.is_always_deserializable() {
            panic!("migration to un-migratable collection: {update:?}\nall updates: {updates:?}");
        }
    }

    // 3. Add a retraction for old version and insertion for new version into updates.
    let next_version = current_version + 1;
    let version_retraction = (version_update_kind(current_version), Diff::MINUS_ONE);
    updates.push(version_retraction);
    let version_insertion = (version_update_kind(next_version), Diff::ONE);
    updates.push(version_insertion);

    // 4. Apply migration to catalog.
    let commit_ts = apply_migration(unopened_catalog_state, updates, commit_ts).await?;

    // 5. Consolidate snapshot to remove old versions.
    unopened_catalog_state.consolidate();

    Ok((next_version, commit_ts))
}

/// Applies one migration's `updates` at `commit_ts` and returns the new upper.
async fn apply_migration(
    unopened_catalog_state: &mut UnopenedPersistCatalogState,
    updates: Vec<(StateUpdateKindJson, Diff)>,
    commit_ts: Timestamp,
) -> Result<Timestamp, CatalogError> {
    if matches!(unopened_catalog_state.mode, Mode::Writable) {
        Ok(unopened_catalog_state
            .compare_and_append(updates, commit_ts)
            .await
            .map_err(|e| e.unwrap_fence_error())?)
    } else {
        let ts = commit_ts;
        let updates = updates
            .into_iter()
            .map(|(kind, diff)| StateUpdate { kind, ts, diff });
        unopened_catalog_state.apply_updates_and_consolidate(updates)?;
        Ok(commit_ts.step_forward())
    }
}

/// Generates a [`proto::StateUpdateKind`] to update the user version.
fn version_update_kind(version: u64) -> StateUpdateKindJson {
    // We can use the current version because Configs can never be migrated and are always wire
    // compatible.
    StateUpdateKind::Config(
        proto::ConfigKey {
            key: USER_VERSION_KEY.to_string(),
        },
        proto::ConfigValue { value: version },
    )
    .into()
}