adapter: support specifying column comments for builtin objects

Signed-off-by: Petros Angelatos <petrosagg@gmail.com>

Author: petrosagg

src/adapter/src/catalog/open.rs

@@ -49,6 +49,7 @@ use mz_sql::catalog::{
     BuiltinsConfig, CatalogError as SqlCatalogError, CatalogItemType, RoleMembership, RoleVars,
 };
 use mz_sql::func::OP_IMPLS;
+use mz_sql::names::CommentObjectId;
 use mz_sql::rbac;
 use mz_sql::session::user::{MZ_SYSTEM_ROLE_ID, SYSTEM_USER};
 use mz_sql::session::vars::{SessionVars, SystemVars, VarError, VarInput};
@@ -727,10 +728,13 @@ fn add_new_remove_old_builtin_items_migration(
             }
         });
     let new_builtin_ids = txn.allocate_system_item_ids(usize_to_u64(new_builtins.len()))?;
-    let new_builtins = new_builtins.into_iter().zip(new_builtin_ids.clone());
+    let new_builtins: Vec<_> = new_builtins
+        .into_iter()
+        .zip(new_builtin_ids.clone())
+        .collect();
 
     // Look for migrated builtins.
-    for (builtin, system_object_mapping, fingerprint) in existing_builtins {
+    for (builtin, system_object_mapping, fingerprint) in existing_builtins.iter().cloned() {
         if system_object_mapping.unique_identifier.fingerprint != fingerprint {
             // `mz_storage_usage_by_shard` cannot be migrated for multiple reasons. Firstly,
             // it was cause the table to be truncated because the contents are not also
@@ -760,7 +764,7 @@ fn add_new_remove_old_builtin_items_migration(
     }
 
     // Add new builtin items to catalog.
-    for ((builtin, fingerprint), (catalog_id, global_id)) in new_builtins {
+    for ((builtin, fingerprint), (catalog_id, global_id)) in new_builtins.iter().cloned() {
         new_builtin_mappings.push(SystemObjectMapping {
             description: SystemObjectDescription {
                 schema_name: builtin.schema().to_string(),
@@ -812,6 +816,85 @@ fn add_new_remove_old_builtin_items_migration(
     }
     txn.set_system_object_mappings(new_builtin_mappings)?;
 
+    // Update comments of all builtin objects
+    let builtins_with_catalog_ids = existing_builtins
+        .iter()
+        .map(|(b, m, _)| (*b, m.unique_identifier.catalog_id))
+        .chain(
+            new_builtins
+                .into_iter()
+                .map(|((b, _), (catalog_id, _))| (b, catalog_id)),
+        );
+    for (builtin, catalog_id) in builtins_with_catalog_ids {
+        match builtin {
+            Builtin::Source(source) => {
+                match source.column_comments {
+                    Some(ref column_comments) => {
+                        assert_eq!(column_comments.len(), source.desc.len());
+                        for (col, &comment) in column_comments.iter().enumerate() {
+                            txn.update_comment(
+                                CommentObjectId::Source(catalog_id),
+                                // Column indices are 1 based
+                                Some(col + 1),
+                                Some(comment.to_owned()),
+                            )?;
+                        }
+                    }
+                    None => {
+                        txn.drop_comments(&BTreeSet::from_iter([CommentObjectId::Source(
+                            catalog_id,
+                        )]))?;
+                    }
+                }
+            }
+            Builtin::View(view) => {
+                match view.column_comments {
+                    Some(ref column_comments) => {
+                        for (col, &comment) in column_comments.iter().enumerate() {
+                            txn.update_comment(
+                                CommentObjectId::View(catalog_id),
+                                // Column indices are 1 based
+                                Some(col + 1),
+                                Some(comment.to_owned()),
+                            )?;
+                        }
+                    }
+                    None => {
+                        txn.drop_comments(&BTreeSet::from_iter([CommentObjectId::View(
+                            catalog_id,
+                        )]))?;
+                    }
+                }
+            }
+            Builtin::Table(table) => {
+                match table.column_comments {
+                    Some(ref column_comments) => {
+                        assert_eq!(column_comments.len(), table.desc.len());
+                        for (col, &comment) in column_comments.iter().enumerate() {
+                            txn.update_comment(
+                                CommentObjectId::Table(catalog_id),
+                                // Column indices are 1 based
+                                Some(col + 1),
+                                Some(comment.to_owned()),
+                            )?;
+                        }
+                    }
+                    None => {
+                        txn.drop_comments(&BTreeSet::from_iter([CommentObjectId::Table(
+                            catalog_id,
+                        )]))?;
+                    }
+                }
+            }
+            Builtin::Log(_)
+            | Builtin::Type(_)
+            | Builtin::Func(_)
+            | Builtin::ContinualTask(_)
+            | Builtin::Index(_)
+            | Builtin::Connection(_) => {}
+        }
+    }
+
     // Anything left in `system_object_mappings` must have been deleted and should be removed from
     // the catalog.
     let mut deleted_system_objects = BTreeSet::new();