evolve item IDs

When running ALTER TABLE, move the table's ID to the new GlobalId,
instead of leaving it at its initial value.

Author: teskje

src/adapter/src/catalog/transact.rs

@@ -766,7 +766,37 @@ impl Catalog {
                 };
                 table.collections.insert(version, new_global_id);
 
-                tx.update_item(id, new_entry.into())?;
+                for use_id in new_entry.referenced_by() {
+                    let mut entry = state.get_entry(use_id).clone();
+                    entry.item = entry.item.replace_item_refs(id, new_global_id);
+                    tx.update_item(*use_id, entry.into())?;
+                }
+
+                let mz_catalog::durable::Item {
+                    id,
+                    oid,
+                    global_id,
+                    schema_id,
+                    name,
+                    create_sql,
+                    owner_id,
+                    privileges,
+                    extra_versions,
+                } = new_entry.into();
+
+                tx.remove_item(id)?;
+                tx.insert_item(
+                    new_global_id,
+                    oid,
+                    global_id,
+                    schema_id,
+                    &name,
+                    create_sql,
+                    owner_id,
+                    privileges,
+                    extra_versions,
+                )?;
+
                 storage_collections_to_register.insert(new_global_id, shard_id);
             }
             Op::CreateDatabase { name, owner_id } => {

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

@@ -5143,7 +5143,7 @@ impl Coordinator {
 
         self.catalog_transact_with_side_effects(Some(ctx), ops, move |coord, _ctx| {
             Box::pin(async move {
-                let entry = coord.catalog().get_entry(&relation_id);
+                let entry = coord.catalog().get_entry(&new_global_id);
                 let CatalogItem::Table(table) = &entry.item else {
                     panic!("programming error, expected table found {:?}", entry.item);
                 };

src/catalog/src/memory/objects.rs

@@ -2009,6 +2009,72 @@ impl CatalogItem {
         }
     }
 
+    pub fn replace_item_refs(&self, old_id: GlobalId, new_id: GlobalId) -> CatalogItem {
+        let do_rewrite = |create_sql: String| -> String {
+            let mut create_stmt = mz_sql::parse::parse(&create_sql)
+                .expect("invalid create sql persisted to catalog")
+                .into_element()
+                .ast;
+            mz_sql::ast::transform::create_stmt_replace_ids(
+                &mut create_stmt,
+                &[(old_id, new_id)].into(),
+            );
+            create_stmt.to_ast_string_stable()
+        };
+
+        match self {
+            CatalogItem::Table(i) => {
+                let mut i = i.clone();
+                i.create_sql = i.create_sql.map(do_rewrite);
+                CatalogItem::Table(i)
+            }
+            CatalogItem::Log(i) => CatalogItem::Log(i.clone()),
+            CatalogItem::Source(i) => {
+                let mut i = i.clone();
+                i.create_sql = i.create_sql.map(do_rewrite);
+                CatalogItem::Source(i)
+            }
+            CatalogItem::Sink(i) => {
+                let mut i = i.clone();
+                i.create_sql = do_rewrite(i.create_sql);
+                CatalogItem::Sink(i)
+            }
+            CatalogItem::View(i) => {
+                let mut i = i.clone();
+                i.create_sql = do_rewrite(i.create_sql);
+                CatalogItem::View(i)
+            }
+            CatalogItem::MaterializedView(i) => {
+                let mut i = i.clone();
+                i.create_sql = do_rewrite(i.create_sql);
+                CatalogItem::MaterializedView(i)
+            }
+            CatalogItem::Index(i) => {
+                let mut i = i.clone();
+                i.create_sql = do_rewrite(i.create_sql);
+                CatalogItem::Index(i)
+            }
+            CatalogItem::Secret(i) => {
+                let mut i = i.clone();
+                i.create_sql = do_rewrite(i.create_sql);
+                CatalogItem::Secret(i)
+            }
+            CatalogItem::Func(_) | CatalogItem::Type(_) => {
+                unimplemented!()
+            }
+            CatalogItem::Connection(i) => {
+                let mut i = i.clone();
+                i.create_sql = do_rewrite(i.create_sql);
+                CatalogItem::Connection(i)
+            }
+            CatalogItem::ContinualTask(i) => {
+                let mut i = i.clone();
+                i.create_sql = do_rewrite(i.create_sql);
+                CatalogItem::ContinualTask(i)
+            }
+        }
+    }
+
     /// Updates the retain history for an item. Returns the previous retain history value. Returns
     /// an error if this item does not support retain history.
     pub fn update_retain_history(

src/sql-parser/src/ast/metadata.rs

@@ -105,6 +105,13 @@ impl RawItemName {
             RawItemName::Id(_, name, _) => name,
         }
     }
+
+    pub fn id_mut(&mut self) -> Option<&mut String> {
+        match self {
+            RawItemName::Name(_) => None,
+            RawItemName::Id(id, _, _) => Some(id),
+        }
+    }
 }
 
 impl AstDisplay for RawItemName {