Merge pull request #27040 from sploiselle/v099-global-id-migration-fix

v.0.99: global id migration fix

Author: Sean Loiselle

src/adapter/src/catalog/apply.rs

@@ -10,8 +10,9 @@
 //! Logic related to applying updates from a [`mz_catalog::durable::DurableCatalogState`] to a
 //! [`CatalogState`].
 
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
 use std::fmt::Debug;
+use std::str::FromStr;
 
 use mz_catalog::builtin::{Builtin, BUILTIN_LOG_LOOKUP, BUILTIN_LOOKUP};
 use mz_catalog::memory::error::{Error, ErrorKind};
@@ -32,9 +33,9 @@ use mz_sql::names::{
     ItemQualifiers, QualifiedItemName, QualifiedSchemaName, ResolvedDatabaseSpecifier, ResolvedIds,
     SchemaSpecifier,
 };
-use mz_sql::rbac;
 use mz_sql::session::user::MZ_SYSTEM_ROLE_ID;
 use mz_sql::session::vars::{VarError, VarInput};
+use mz_sql::{plan, rbac};
 use mz_sql_parser::ast::Expr;
 use mz_storage_types::sources::Timeline;
 use once_cell::sync::Lazy;
@@ -46,6 +47,8 @@ use crate::AdapterError;
 
 enum ApplyUpdateError {
     Error(Error),
+    AwaitingIdDependency((GlobalId, mz_catalog::durable::Item, Diff)),
+    AwaitingNameDependency((String, mz_catalog::durable::Item, Diff)),
 }
 
 impl CatalogState {
@@ -66,13 +69,126 @@ impl CatalogState {
         &mut self,
         updates: Vec<StateUpdate>,
     ) -> Result<(), Error> {
-        for StateUpdate { kind, diff } in updates {
+        // TODO(v100): we can refactor this to assume that items are
+        // topologically sorted by their `GlobalId`.
+        let mut awaiting_id_dependencies: BTreeMap<GlobalId, Vec<_>> = BTreeMap::new();
+        let mut awaiting_name_dependencies: BTreeMap<String, Vec<_>> = BTreeMap::new();
+        let mut updates: VecDeque<_> = updates.into_iter().collect();
+        while let Some(StateUpdate { kind, diff }) = updates.pop_front() {
             assert_eq!(
                 diff, 1,
                 "initial catalog updates should be consolidated: ({kind:?}, {diff:?})"
             );
-            self.apply_update(kind, diff)
-                .map_err(|ApplyUpdateError::Error(err)| err)?;
+            match self.apply_update(kind, diff) {
+                Ok(None) => {}
+                Ok(Some(id)) => {
+                    // Enqueue any items waiting on this dependency.
+                    let mut resolved_dependent_items = Vec::new();
+                    if let Some(dependent_items) = awaiting_id_dependencies.remove(&id) {
+                        resolved_dependent_items.extend(dependent_items);
+                    }
+                    let entry = self.get_entry(&id);
+                    let full_name = self.resolve_full_name(entry.name(), None);
+                    if let Some(dependent_items) =
+                        awaiting_name_dependencies.remove(&full_name.to_string())
+                    {
+                        resolved_dependent_items.extend(dependent_items);
+                    }
+                    let resolved_dependent_items =
+                        resolved_dependent_items
+                            .into_iter()
+                            .map(|(item, diff)| StateUpdate {
+                                kind: StateUpdateKind::Item(item),
+                                diff,
+                            });
+                    updates.extend(resolved_dependent_items);
+                }
+                Err(ApplyUpdateError::Error(err)) => return Err(err),
+                Err(ApplyUpdateError::AwaitingIdDependency((id, item, diff))) => {
+                    awaiting_id_dependencies
+                        .entry(id)
+                        .or_default()
+                        .push((item, diff));
+                }
+                Err(ApplyUpdateError::AwaitingNameDependency((name, item, diff))) => {
+                    awaiting_name_dependencies
+                        .entry(name)
+                        .or_default()
+                        .push((item, diff));
+                }
+            }
+        }
+
+        // Error on any unsatisfied dependencies.
+        if let Some((missing_dep, mut dependents)) = awaiting_id_dependencies.into_iter().next() {
+            let (
+                mz_catalog::durable::Item {
+                    id,
+                    oid: _,
+                    schema_id,
+                    name,
+                    create_sql: _,
+                    owner_id: _,
+                    privileges: _,
+                },
+                diff,
+            ) = dependents.remove(0);
+            let schema = self.find_non_temp_schema(&schema_id);
+            let name = QualifiedItemName {
+                qualifiers: ItemQualifiers {
+                    database_spec: schema.database().clone(),
+                    schema_spec: schema.id().clone(),
+                },
+                item: name,
+            };
+            let name = self.resolve_full_name(&name, None);
+            let action = if diff == 1 { "deserialize" } else { "remove" };
+
+            return Err(Error::new(ErrorKind::Corruption {
+                detail: format!(
+                    "failed to {} item {} ({}): {}",
+                    action,
+                    id,
+                    name,
+                    plan::PlanError::InvalidId(missing_dep)
+                ),
+            }));
+        }
+
+        if let Some((missing_dep, mut dependents)) = awaiting_name_dependencies.into_iter().next() {
+            let (
+                mz_catalog::durable::Item {
+                    id,
+                    oid: _,
+                    schema_id,
+                    name,
+                    create_sql: _,
+                    owner_id: _,
+                    privileges: _,
+                },
+                diff,
+            ) = dependents.remove(0);
+            let schema = self.find_non_temp_schema(&schema_id);
+            let name = QualifiedItemName {
+                qualifiers: ItemQualifiers {
+                    database_spec: schema.database().clone(),
+                    schema_spec: schema.id().clone(),
+                },
+                item: name,
+            };
+            let name = self.resolve_full_name(&name, None);
+            let action = if diff == 1 { "deserialize" } else { "remove" };
+            return Err(Error::new(ErrorKind::Corruption {
+                detail: format!(
+                    "failed to {} item {} ({}): {}",
+                    action,
+                    id,
+                    name,
+                    Error {
+                        kind: ErrorKind::Sql(SqlCatalogError::UnknownItem(missing_dep))
+                    }
+                ),
+            }));
         }
 
         Ok(())
@@ -615,12 +731,18 @@ impl CatalogState {
     /// Applies an item update to `self`.
     ///
     /// Returns a `GlobalId` on success, if the applied update added a new `GlobalID` to `self`.
+    /// Returns a dependency on failure, if the update could not be applied due to a missing
+    /// dependency.
     #[instrument(level = "debug")]
     fn apply_item_update(
         &mut self,
         item: mz_catalog::durable::Item,
         diff: Diff,
     ) -> Result<Option<GlobalId>, ApplyUpdateError> {
+        // If we knew beforehand that the items were being applied in dependency
+        // order, then we could fully delegate to `self.insert_item(...)` and`self.drop_item(...)`.
+        // However, we don't know that items are applied in dependency order, so we must handle the
+        // case that the item is valid, but we haven't applied all of its dependencies yet.
         match diff {
             1 => {
                 // TODO(benesch): a better way of detecting when a view has depended
@@ -641,6 +763,27 @@ impl CatalogState {
                             },
                         )));
                     }
+                    // If we were missing a dependency, wait for it to be added.
+                    Err(AdapterError::PlanError(plan::PlanError::InvalidId(missing_dep))) => {
+                        return Err(ApplyUpdateError::AwaitingIdDependency((
+                            missing_dep,
+                            item,
+                            diff,
+                        )));
+                    }
+                    // If we were missing a dependency, wait for it to be added.
+                    Err(AdapterError::PlanError(plan::PlanError::Catalog(
+                        SqlCatalogError::UnknownItem(missing_dep),
+                    ))) => {
+                        return match GlobalId::from_str(&missing_dep) {
+                            Ok(id) => Err(ApplyUpdateError::AwaitingIdDependency((id, item, diff))),
+                            Err(_) => Err(ApplyUpdateError::AwaitingNameDependency((
+                                missing_dep,
+                                item,
+                                diff,
+                            ))),
+                        }
+                    }
                     Err(e) => {
                         let schema = self.find_non_temp_schema(&item.schema_id);
                         let name = QualifiedItemName {
@@ -678,6 +821,13 @@ impl CatalogState {
                 Ok(Some(item.id))
             }
             -1 => {
+                let entry = self.get_entry(&item.id);
+                if let Some(id) = entry.referenced_by().first() {
+                    return Err(ApplyUpdateError::AwaitingIdDependency((*id, item, diff)));
+                }
+                if let Some(id) = entry.used_by().first() {
+                    return Err(ApplyUpdateError::AwaitingIdDependency((*id, item, diff)));
+                }
                 self.drop_item(item.id);
                 Ok(None)
             }

src/adapter/src/catalog/migrate.rs

@@ -83,7 +83,7 @@ where
 pub(crate) async fn migrate(
     state: &CatalogState,
     tx: &mut Transaction<'_>,
-    _now: NowFn,
+    now: NowFn,
     _boot_ts: Timestamp,
     _connection_context: &ConnectionContext,
 ) -> Result<(), anyhow::Error> {
@@ -158,6 +158,7 @@ pub(crate) async fn migrate(
     // Each migration should be a function that takes `tx` and `conn_cat` as
     // input and stages arbitrary transformations to the catalog on `tx`.
     mysql_subsources_remove_unnecessary_db_name(tx, &conn_cat)?;
+    fix_dependency_order_0_99_1(tx, &conn_cat, now)?;
 
     info!(
         "migration from catalog version {:?} complete",
@@ -178,7 +179,7 @@ pub(crate) async fn migrate(
 ///   `GlobalId`s they wish to reassign to `needs_new_id`.
 /// - Assumes all items in `needs_new_id` are present in `conn_catalog` and that
 ///   their types do not change.
-fn _assign_new_user_global_ids(
+fn assign_new_user_global_ids(
     tx: &mut Transaction<'_>,
     conn_catalog: &ConnCatalog,
     now: NowFn,
@@ -260,7 +261,7 @@ fn _assign_new_user_global_ids(
 
         let object_type = entry.item_type().into();
 
-        _add_to_audit_log(
+        add_to_audit_log(
             tx,
             mz_audit_log::EventType::Create,
             object_type,
@@ -271,7 +272,7 @@ fn _assign_new_user_global_ids(
             occurred_at,
         )?;
 
-        _add_to_audit_log(
+        add_to_audit_log(
             tx,
             mz_audit_log::EventType::Drop,
             object_type,
@@ -416,7 +417,7 @@ fn _assign_new_user_global_ids(
 // Please include the adapter team on any code reviews that add or edit
 // migrations.
 
-fn _add_to_audit_log(
+fn add_to_audit_log(
     tx: &mut Transaction,
     event_type: mz_audit_log::EventType,
     object_type: mz_audit_log::ObjectType,
@@ -646,6 +647,65 @@ fn mysql_subsources_remove_unnecessary_db_name(
     Ok(())
 }
 
+/// In v0.98.x, we inverted the dependency order of sources + subsources and
+/// migrated those items, as well as their dependencies, to new IDs.
+/// Unfortunately, because of a bug, this did not place items in the correct
+/// order and it was possible that some items depend on items with IDs less than
+/// its own.
+///
+/// This PR goes back and fixes any IDs with that broken relationship.
+fn fix_dependency_order_0_99_1(
+    tx: &mut Transaction<'_>,
+    conn_catalog: &ConnCatalog,
+    now: NowFn,
+) -> Result<(), anyhow::Error> {
+    // A vector in the _new_ dependency order. Because the `vec` doesn't have
+    // any built-in de-duplication, we will need to deduplicate these values
+    // elsewhere.
+    let mut needs_new_id = vec![];
+
+    for item in conn_catalog.get_items() {
+        for dependent_id in item.used_by() {
+            // If an item has a dependent with an ID less than its own, that
+            // dependent needs an ID that will be pushed forward ahead of this
+            // ID.
+            if *dependent_id < item.id() {
+                needs_new_id.push(*dependent_id);
+            }
+        }
+    }
+
+    let mut remaining_updates = std::collections::VecDeque::from_iter(needs_new_id.drain(..));
+
+    let state = conn_catalog.state();
+
+    // If this item's ID must be moved forward, so too must every item that
+    // depends on it except for its primary source ID.
+    while let Some(id) = remaining_updates.pop_front() {
+        needs_new_id.push(id);
+        // If we push this item forward, we must all push forward all IDs that
+        // refer to this ID (i.e. we need to ensure that all of an item's
+        // dependents have IDs greater than its own).
+        remaining_updates.extend(state.get_entry(&id).used_by().iter().cloned());
+    }
+
+    // Ensure that each ID is present only once and that it is in the greatest
+    // position.
+    let mut id_dedup = BTreeSet::new();
+    needs_new_id = needs_new_id
+        .into_iter()
+        .rev()
+        .filter(|id| id_dedup.insert(*id))
+        .collect();
+
+    // Flip this back around in the right order.
+    needs_new_id = needs_new_id.into_iter().rev().collect();
+
+    assign_new_user_global_ids(tx, conn_catalog, now, needs_new_id)?;
+
+    Ok(())
+}
+
 // Durable migrations
 
 /// Migrations that run only on the durable catalog before any data is loaded into memory.

src/adapter/src/coord.rs

@@ -1499,13 +1499,13 @@ impl Coordinator {
                 // beneficial.
                 !entry.id().is_user()
                     || entry
-                        .used_by()
+                        .uses()
                         .iter()
-                        .all(|dependent_id| *dependent_id > entry.id),
-                "user item dependencies should respect `GlobalId`'s PartialOrd \
-                but {:?} depends on {:?}",
+                        .all(|dependency_id| *dependency_id < entry.id),
+                "entries should only use to items with lesser `GlobalId`s, but \
+                but {:?} uses {:?}",
                 entry.id,
-                entry.used_by()
+                entry.uses()
             );
 
             debug!(