WIP

Author: jkosh44

src/adapter/src/catalog.rs

@@ -718,13 +718,14 @@ impl Catalog {
     #[cfg(test)]
     pub async fn allocate_system_id(&self) -> Result<(CatalogItemId, GlobalId), Error> {
         use mz_ore::collections::CollectionExt;
-        self.storage()
-            .await
-            .allocate_system_ids(1)
-            .await
-            .maybe_terminate("allocating system ids")
-            .map(|ids| ids.into_element())
-            .err_into()
+
+        let mut txn = self.storage().await.transaction().await?;
+        let id = txn
+            .allocate_system_item_ids(1)
+            .maybe_terminate("allocating system ids")?
+            .into_element();
+        txn.commit().await?;
+        Ok(id)
     }
 
     /// Get the next system item ID without allocating it.

src/adapter/src/catalog/open.rs

@@ -28,9 +28,7 @@ use mz_catalog::config::{ClusterReplicaSizeMap, StateConfig};
 use mz_catalog::durable::objects::{
     SystemObjectDescription, SystemObjectMapping, SystemObjectUniqueIdentifier,
 };
-use mz_catalog::durable::{
-    ClusterVariant, ClusterVariantManaged, Transaction, SYSTEM_CLUSTER_ID_ALLOC_KEY,
-};
+use mz_catalog::durable::{ClusterVariant, ClusterVariantManaged, Transaction};
 use mz_catalog::expr_cache::{
     ExpressionCacheConfig, ExpressionCacheHandle, GlobalExpressions, LocalExpressions,
 };
@@ -626,6 +624,8 @@ impl Catalog {
                 );
             }
 
+            catalog.storage().await.bootstrap_complete();
+
             Ok(OpenCatalogResult {
                 catalog,
                 storage_collections_to_drop,
@@ -766,6 +766,10 @@ impl Catalog {
 
             let (new_item_id, new_global_id) = match id {
                 CatalogItemId::System(_) => txn.allocate_system_item_ids(1)?.into_element(),
+                CatalogItemId::IntrospectionSourceIndex(id) => (
+                    CatalogItemId::IntrospectionSourceIndex(id),
+                    GlobalId::IntrospectionSourceIndex(id),
+                ),
                 CatalogItemId::User(_) => txn.allocate_user_item_ids(1)?.into_element(),
                 _ => unreachable!("can't migrate id: {id}"),
             };
@@ -1216,10 +1220,7 @@ fn add_new_builtin_clusters_migration(
         if !cluster_names.contains(builtin_cluster.name) {
             let cluster_size = builtin_cluster_sizes.get_size(builtin_cluster.name)?;
             let cluster_allocation = cluster_sizes.get_allocation_by_name(&cluster_size)?;
-            let id = txn.get_and_increment_id(SYSTEM_CLUSTER_ID_ALLOC_KEY.to_string())?;
-            let id = ClusterId::System(id);
             txn.insert_system_cluster(
-                id,
                 builtin_cluster.name,
                 vec![],
                 builtin_cluster.privileges.to_vec(),
@@ -1259,13 +1260,9 @@ fn add_new_remove_old_builtin_introspection_source_migration(
             }
         }
 
-        let new_ids = txn.allocate_system_item_ids(usize_to_u64(new_logs.len()))?;
-        let new_entries = new_logs
-            .into_iter()
-            .zip_eq(new_ids)
-            .map(|(log, (item_id, gid))| (log, item_id, gid));
-
-        for (log, item_id, gid) in new_entries {
+        for log in new_logs {
+            let (item_id, gid) =
+                Transaction::allocate_introspection_source_index_id(&cluster.id, log.variant);
             new_indexes.push((cluster.id, log.name.to_string(), item_id, gid));
         }
 

src/adapter/src/catalog/transact.rs

@@ -32,7 +32,6 @@ use mz_catalog::memory::objects::{
 use mz_catalog::SYSTEM_CONN_ID;
 use mz_controller::clusters::{ManagedReplicaLocation, ReplicaConfig, ReplicaLocation};
 use mz_controller_types::{ClusterId, ReplicaId};
-use mz_ore::cast::usize_to_u64;
 use mz_ore::collections::HashSet;
 use mz_ore::instrument;
 use mz_ore::now::EpochMillis;
@@ -56,6 +55,7 @@ use mz_sql::session::vars::{Value as VarValue, VarInput};
 use mz_sql::{rbac, DEFAULT_SCHEMA};
 use mz_sql_parser::ast::{QualifiedReplica, Value};
 use mz_storage_client::controller::StorageController;
+use timely::Container;
 use tracing::{info, trace};
 
 use crate::catalog::{
@@ -903,12 +903,19 @@ impl Catalog {
                 let privileges: Vec<_> =
                     merge_mz_acl_items(owner_privileges.into_iter().chain(default_privileges))
                         .collect();
-                let introspection_source_ids =
-                    tx.allocate_system_item_ids(usize_to_u64(introspection_sources.len()))?;
+                let introspection_source_ids: Vec<_> = introspection_sources
+                    .iter()
+                    .map(|introspection_source| {
+                        Transaction::allocate_introspection_source_index_id(
+                            &id,
+                            introspection_source.variant,
+                        )
+                    })
+                    .collect();
 
                 let introspection_sources = introspection_sources
                     .into_iter()
-                    .zip_eq(introspection_source_ids.into_iter())
+                    .zip_eq(introspection_source_ids)
                     .map(|(log, (item_id, gid))| (log, item_id, gid))
                     .collect();
 

src/adapter/src/coord.rs

@@ -2301,7 +2301,8 @@ impl Coordinator {
                             let id_too_large = match id {
                                 CatalogItemId::System(id) => *id >= next_system_item_id,
                                 CatalogItemId::User(id) => *id >= next_user_item_id,
-                                CatalogItemId::Transient(_) => false,
+                                CatalogItemId::IntrospectionSourceIndex(_)
+                                | CatalogItemId::Transient(_) => false,
                             };
                             if id_too_large {
                                 info!(

src/catalog/protos/hashes.json

@@ -1,7 +1,7 @@
 [
   {
     "name": "objects.proto",
-    "md5": "2d781c72c4a56b13dfb1b4215f3614f0"
+    "md5": "4db86834a7af4b5a310e4a8a1ba92e0e"
   },
   {
     "name": "objects_v67.proto",
@@ -26,5 +26,9 @@
   {
     "name": "objects_v72.proto",
     "md5": "b21cb2b1b41649c78405731e53560d59"
+  },
+  {
+    "name": "objects_v73.proto",
+    "md5": "65c222651e901fac2391423a4f034fcc"
   }
 ]

src/catalog/protos/objects.proto

@@ -83,7 +83,7 @@ message ClusterIntrospectionSourceIndexValue {
   // doesn't know how to read the new SystemCatalogItemId type.
   uint64 index_id = 1;
   uint32 oid = 2;
-  SystemGlobalId global_id = 3;
+  IntrospectionSourceIndexGlobalId global_id = 3;
 }
 
 message ClusterReplicaKey {
@@ -307,6 +307,7 @@ message CatalogItemId {
     uint64 system = 1;
     uint64 user = 2;
     uint64 transient = 3;
+    uint64 introspection_source_index = 4;
   }
 }
 
@@ -321,6 +322,7 @@ message GlobalId {
     uint64 user = 2;
     uint64 transient = 3;
     Empty explain = 4;
+    uint64 introspection_source_index = 5;
   }
 }
 
@@ -329,10 +331,15 @@ message SystemGlobalId {
   uint64 value = 1;
 }
 
+/// A newtype wrapper for a `GlobalId` that is always in the "introspection source index" namespace.
+message IntrospectionSourceIndexGlobalId {
+  uint64 value = 1;
+}
+
 message ClusterId {
   oneof value {
-    uint64 system = 1;
-    uint64 user = 2;
+    uint32 system = 1;
+    uint32 user = 2;
   }
 }