Serve /metrics/custom from the catalog (2/3)

doc/user/content/reference/system-catalog/mz_catalog.md

@@ -703,4 +703,6 @@ Field          | Type                 | Meaning
 [`uint8 list`]: /sql/types/list
 [`uint4`]: /sql/types/uint4
 
+<!-- RELATION_SPEC_UNDOCUMENTED mz_catalog.mz_apis -->
+<!-- RELATION_SPEC_UNDOCUMENTED mz_catalog.mz_metrics -->
 <!-- RELATION_SPEC_UNDOCUMENTED mz_catalog.mz_operators -->

src/adapter/src/catalog.rs

@@ -1454,6 +1454,8 @@ pub(crate) fn comment_id_to_audit_object_type(id: CommentObjectId) -> ObjectType
         CommentObjectId::Connection(_) => ObjectType::Connection,
         CommentObjectId::Type(_) => ObjectType::Type,
         CommentObjectId::Secret(_) => ObjectType::Secret,
+        CommentObjectId::Api(_) => ObjectType::Api,
+        CommentObjectId::Metric(_) => ObjectType::Metric,
         CommentObjectId::Role(_) => ObjectType::Role,
         CommentObjectId::Database(_) => ObjectType::Database,
         CommentObjectId::Schema(_) => ObjectType::Schema,
@@ -1490,6 +1492,8 @@ pub(crate) fn system_object_type_to_audit_object_type(
             mz_sql::catalog::ObjectType::Schema => ObjectType::Schema,
             mz_sql::catalog::ObjectType::Func => ObjectType::Func,
             mz_sql::catalog::ObjectType::NetworkPolicy => ObjectType::NetworkPolicy,
+            mz_sql::catalog::ObjectType::Api => ObjectType::Api,
+            mz_sql::catalog::ObjectType::Metric => ObjectType::Metric,
         },
         SystemObjectType::System => ObjectType::System,
     }

src/adapter/src/catalog/apply.rs

@@ -2337,7 +2337,9 @@ fn sort_updates(updates: Vec<StateUpdate>) -> Vec<StateUpdate> {
                 CatalogItemType::Table => tables.push(update),
                 CatalogItemType::View
                 | CatalogItemType::MaterializedView
-                | CatalogItemType::Index => derived_items.push(update),
+                | CatalogItemType::Index
+                | CatalogItemType::Api
+                | CatalogItemType::Metric => derived_items.push(update),
                 CatalogItemType::Sink => sinks.push(update),
             }
         }
@@ -2402,7 +2404,9 @@ fn sort_updates(updates: Vec<StateUpdate>) -> Vec<StateUpdate> {
                 CatalogItemType::Table => tables.push(update),
                 CatalogItemType::View
                 | CatalogItemType::MaterializedView
-                | CatalogItemType::Index => derived_items.push(update),
+                | CatalogItemType::Index
+                | CatalogItemType::Api
+                | CatalogItemType::Metric => derived_items.push(update),
                 CatalogItemType::Sink => sinks.push(update),
             }
         }

src/adapter/src/catalog/builtin_table_updates.rs

@@ -607,6 +607,8 @@ impl CatalogState {
             CatalogItem::Connection(connection) => {
                 self.pack_connection_update(id, connection, diff)
             }
+            // TODO: emit mz_apis / mz_metrics rows once those builtin tables exist.
+            CatalogItem::Api(_) | CatalogItem::Metric(_) => vec![],
         };
 
         if !entry.item().is_temporary() {
@@ -1869,7 +1871,9 @@ impl CatalogState {
             | CommentObjectId::Func(global_id)
             | CommentObjectId::Connection(global_id)
             | CommentObjectId::Secret(global_id)
-            | CommentObjectId::Type(global_id) => global_id.to_string(),
+            | CommentObjectId::Type(global_id)
+            | CommentObjectId::Api(global_id)
+            | CommentObjectId::Metric(global_id) => global_id.to_string(),
             CommentObjectId::Role(role_id) => role_id.to_string(),
             CommentObjectId::Database(database_id) => database_id.to_string(),
             CommentObjectId::Schema((_, schema_id)) => schema_id.to_string(),

src/adapter/src/catalog/consistency.rs

@@ -273,7 +273,9 @@ impl CatalogState {
                 | CommentObjectId::Func(item_id)
                 | CommentObjectId::Connection(item_id)
                 | CommentObjectId::Type(item_id)
-                | CommentObjectId::Secret(item_id) => {
+                | CommentObjectId::Secret(item_id)
+                | CommentObjectId::Api(item_id)
+                | CommentObjectId::Metric(item_id) => {
                     let entry = self.entry_by_id.get(&item_id);
                     match entry {
                         None => comment_inconsistencies

src/adapter/src/catalog/open.rs

@@ -918,7 +918,9 @@ fn add_new_remove_old_builtin_items_migration(
             | CatalogItemType::Type
             | CatalogItemType::Func
             | CatalogItemType::Secret
-            | CatalogItemType::Connection => continue,
+            | CatalogItemType::Connection
+            | CatalogItemType::Api
+            | CatalogItemType::Metric => continue,
         };
         deleted_comments.insert(comment_id);
     }

src/adapter/src/catalog/state.rs

@@ -30,8 +30,8 @@ use mz_catalog::config::{AwsPrincipalContext, ClusterReplicaSizeMap};
 use mz_catalog::expr_cache::LocalExpressions;
 use mz_catalog::memory::error::{Error, ErrorKind};
 use mz_catalog::memory::objects::{
-    CatalogCollectionEntry, CatalogEntry, CatalogItem, Cluster, ClusterReplica, CommentsMap,
-    Connection, DataSourceDesc, Database, DefaultPrivileges, Index, MaterializedView,
+    Api, CatalogCollectionEntry, CatalogEntry, CatalogItem, Cluster, ClusterReplica, CommentsMap,
+    Connection, DataSourceDesc, Database, DefaultPrivileges, Index, MaterializedView, Metric,
     NetworkPolicy, Role, RoleAuth, Schema, Secret, Sink, Source, SourceReferences, Table,
     TableDataSource, Type, View,
 };
@@ -448,11 +448,15 @@ impl CatalogState {
                 let on_item_id = self.get_entry_by_global_id(&idx.on).id();
                 self.introspection_dependencies_inner(on_item_id, out)
             }
+            CatalogItem::Metric(metric) => {
+                self.introspection_dependencies_inner(metric.values_from, out)
+            }
             CatalogItem::Table(_)
             | CatalogItem::Source(_)
             | CatalogItem::Type(_)
             | CatalogItem::Func(_)
-            | CatalogItem::Secret(_) => (),
+            | CatalogItem::Secret(_)
+            | CatalogItem::Api(_) => (),
         }
     }
 
@@ -1516,6 +1520,24 @@ impl CatalogState {
                 details,
                 resolved_ids,
             }),
+            Plan::CreateApi(mz_sql::plan::CreateApiPlan { api, .. }) => CatalogItem::Api(Api {
+                create_sql: api.create_sql,
+                global_id,
+                cluster_id: api.cluster_id,
+                resolved_ids,
+            }),
+            Plan::CreateMetric(mz_sql::plan::CreateMetricPlan { metric, .. }) => {
+                CatalogItem::Metric(Metric {
+                    create_sql: metric.create_sql,
+                    global_id,
+                    api_id: metric.api_id,
+                    metric_type: metric.metric_type,
+                    help: metric.help,
+                    values_from: metric.values_from,
+                    value_column: metric.value_column,
+                    resolved_ids,
+                })
+            }
             _ => {
                 return Err((
                     Error::new(ErrorKind::Corruption {
@@ -1883,7 +1905,9 @@ impl CatalogState {
             | CatalogItemType::MaterializedView
             | CatalogItemType::Index
             | CatalogItemType::Secret
-            | CatalogItemType::Connection => schema.items[builtin.name()],
+            | CatalogItemType::Connection
+            | CatalogItemType::Api
+            | CatalogItemType::Metric => schema.items[builtin.name()],
         }
     }
 
@@ -2697,7 +2721,9 @@ impl CatalogState {
             | CommentObjectId::Func(id)
             | CommentObjectId::Connection(id)
             | CommentObjectId::Type(id)
-            | CommentObjectId::Secret(id) => Some(*id),
+            | CommentObjectId::Secret(id)
+            | CommentObjectId::Api(id)
+            | CommentObjectId::Metric(id) => Some(*id),
             CommentObjectId::Role(_)
             | CommentObjectId::Database(_)
             | CommentObjectId::Schema(_)
@@ -2726,7 +2752,9 @@ impl CatalogState {
             | CommentObjectId::Func(id)
             | CommentObjectId::Connection(id)
             | CommentObjectId::Type(id)
-            | CommentObjectId::Secret(id) => {
+            | CommentObjectId::Secret(id)
+            | CommentObjectId::Api(id)
+            | CommentObjectId::Metric(id) => {
                 let item = self.get_entry(&id);
                 let name = self.resolve_full_name(item.name(), Some(conn_id));
                 name.to_string()

src/adapter/src/catalog/timeline.rs

@@ -97,7 +97,9 @@ impl Catalog {
                         | CatalogItem::Func(_)
                         | CatalogItem::Secret(_)
                         | CatalogItem::Connection(_)
-                        | CatalogItem::Log(_) => {}
+                        | CatalogItem::Log(_)
+                        | CatalogItem::Api(_)
+                        | CatalogItem::Metric(_) => {}
                     }
                 }
             }
@@ -233,7 +235,9 @@ impl Catalog {
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)
-                    | CatalogItem::Connection(_) => {}
+                    | CatalogItem::Connection(_)
+                    | CatalogItem::Api(_)
+                    | CatalogItem::Metric(_) => {}
                 }
             }
         }

src/adapter/src/catalog/transact.rs

@@ -606,7 +606,9 @@ impl Catalog {
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)
-                    | CatalogItem::Connection(_) => {}
+                    | CatalogItem::Connection(_)
+                    | CatalogItem::Api(_)
+                    | CatalogItem::Metric(_) => {}
                 }
             }
         }
@@ -1396,7 +1398,9 @@ impl Catalog {
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)
-                    | CatalogItem::Connection(_) => (),
+                    | CatalogItem::Connection(_)
+                    | CatalogItem::Api(_)
+                    | CatalogItem::Metric(_) => (),
                 }
 
                 let system_user = session.map_or(false, |s| s.user().is_system_user());
@@ -1580,7 +1584,9 @@ impl Catalog {
                         | CatalogItem::Type(_)
                         | CatalogItem::Func(_)
                         | CatalogItem::Secret(_)
-                        | CatalogItem::Connection(_) => EventDetails::IdFullNameV1(IdFullNameV1 {
+                        | CatalogItem::Connection(_)
+                        | CatalogItem::Api(_)
+                        | CatalogItem::Metric(_) => EventDetails::IdFullNameV1(IdFullNameV1 {
                             id: id.to_string(),
                             name,
                         }),

src/adapter/src/command.rs

@@ -583,6 +583,10 @@ pub enum ExecuteResponse {
     CreatedType,
     /// The requested network policy was created.
     CreatedNetworkPolicy,
+    /// The requested API was created.
+    CreatedApi,
+    /// The requested metric was created.
+    CreatedMetric,
     /// The requested prepared statement was removed.
     Deallocate { all: bool },
     /// The requested cursor was declared.
@@ -749,6 +753,8 @@ impl TryInto<ExecuteResponse> for ExecuteResponseKind {
                 Ok(ExecuteResponse::CreatedMaterializedView)
             }
             ExecuteResponseKind::CreatedNetworkPolicy => Ok(ExecuteResponse::CreatedNetworkPolicy),
+            ExecuteResponseKind::CreatedApi => Ok(ExecuteResponse::CreatedApi),
+            ExecuteResponseKind::CreatedMetric => Ok(ExecuteResponse::CreatedMetric),
             ExecuteResponseKind::CreatedType => Ok(ExecuteResponse::CreatedType),
             ExecuteResponseKind::Deallocate => Err(()),
             ExecuteResponseKind::DeclaredCursor => Ok(ExecuteResponse::DeclaredCursor),
@@ -812,6 +818,8 @@ impl ExecuteResponse {
             CreatedMaterializedView { .. } => Some("CREATE MATERIALIZED VIEW".into()),
             CreatedType => Some("CREATE TYPE".into()),
             CreatedNetworkPolicy => Some("CREATE NETWORKPOLICY".into()),
+            CreatedApi => Some("CREATE API".into()),
+            CreatedMetric => Some("CREATE METRIC".into()),
             Deallocate { all } => Some(format!("DEALLOCATE{}", if *all { " ALL" } else { "" })),
             DeclaredCursor => Some("DECLARE CURSOR".into()),
             Deleted(n) => Some(format!("DELETE {}", n)),
@@ -905,6 +913,8 @@ impl ExecuteResponse {
             CreateType => &[CreatedType],
             PlanKind::Deallocate => &[ExecuteResponseKind::Deallocate],
             CreateNetworkPolicy => &[CreatedNetworkPolicy],
+            CreateApi => &[CreatedApi],
+            CreateMetric => &[CreatedMetric],
             Declare => &[DeclaredCursor],
             DiscardTemp => &[DiscardedTemp],
             DiscardAll => &[DiscardedAll],

src/adapter/src/coord.rs

@@ -2352,7 +2352,9 @@ impl Coordinator {
                 CatalogItem::Log(_)
                 | CatalogItem::Type(_)
                 | CatalogItem::Func(_)
-                | CatalogItem::Secret(_) => {}
+                | CatalogItem::Secret(_)
+                | CatalogItem::Api(_)
+                | CatalogItem::Metric(_) => {}
             }
         }
 
@@ -3020,7 +3022,9 @@ impl Coordinator {
                 | CatalogItem::Type(_)
                 | CatalogItem::Func(_)
                 | CatalogItem::Secret(_)
-                | CatalogItem::Connection(_) => (),
+                | CatalogItem::Connection(_)
+                | CatalogItem::Api(_)
+                | CatalogItem::Metric(_) => (),
             }
         }
 
@@ -3411,7 +3415,9 @@ impl Coordinator {
                 | CatalogItem::Type(_)
                 | CatalogItem::Func(_)
                 | CatalogItem::Secret(_)
-                | CatalogItem::Connection(_) => (),
+                | CatalogItem::Connection(_)
+                | CatalogItem::Api(_)
+                | CatalogItem::Metric(_) => (),
             }
         }
 
@@ -3443,7 +3449,9 @@ impl Coordinator {
                 | CatalogItem::Type(_)
                 | CatalogItem::Func(_)
                 | CatalogItem::Secret(_)
-                | CatalogItem::Connection(_) => continue,
+                | CatalogItem::Connection(_)
+                | CatalogItem::Api(_)
+                | CatalogItem::Metric(_) => continue,
             };
             if let Some(plan) = self.catalog.try_get_physical_plan(&gid) {
                 catalog_ids.push(gid);

src/adapter/src/coord/appends.rs

@@ -971,6 +971,8 @@ pub(crate) fn waiting_on_startup_appends(
         | Plan::CreateSchema(_)
         | Plan::CreateRole(_)
         | Plan::CreateNetworkPolicy(_)
+        | Plan::CreateApi(_)
+        | Plan::CreateMetric(_)
         | Plan::CreateCluster(_)
         | Plan::CreateClusterReplica(_)
         | Plan::CreateSource(_)

src/adapter/src/coord/catalog_implications.rs

@@ -1551,7 +1551,9 @@ impl Coordinator {
                     | CatalogItem::Index(_)
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
-                    | CatalogItem::Secret(_) => {
+                    | CatalogItem::Secret(_)
+                    | CatalogItem::Api(_)
+                    | CatalogItem::Metric(_) => {
                         // Other item types don't have connection dependencies
                         // that need updating.
                     }
@@ -1762,6 +1764,8 @@ impl CatalogImplication {
                 CatalogItem::Log(_) => {}
                 CatalogItem::Type(_) => {}
                 CatalogItem::Func(_) => {}
+                CatalogItem::Api(_) => {}
+                CatalogItem::Metric(_) => {}
             },
             ParsedStateUpdateKind::TemporaryItem {
                 durable_item: _,
@@ -1801,6 +1805,8 @@ impl CatalogImplication {
                 CatalogItem::Log(_) => {}
                 CatalogItem::Type(_) => {}
                 CatalogItem::Func(_) => {}
+                CatalogItem::Api(_) => {}
+                CatalogItem::Metric(_) => {}
             },
             ParsedStateUpdateKind::Cluster {
                 durable_cluster: _,

src/adapter/src/coord/catalog_serving.rs

@@ -82,6 +82,8 @@ pub fn auto_run_on_catalog_server<'a, 's, 'p>(
         | Plan::CreateSchema(_)
         | Plan::CreateRole(_)
         | Plan::CreateNetworkPolicy(_)
+        | Plan::CreateApi(_)
+        | Plan::CreateMetric(_)
         | Plan::CreateCluster(_)
         | Plan::CreateClusterReplica(_)
         | Plan::CreateSource(_)

src/adapter/src/coord/command_handler.rs

@@ -1312,6 +1312,8 @@ impl Coordinator {
                     | Statement::CreateView(_)
                     | Statement::CreateWebhookSource(_)
                     | Statement::CreateNetworkPolicy(_)
+                    | Statement::CreateApi(_)
+                    | Statement::CreateMetric(_)
                     | Statement::Delete(_)
                     | Statement::DropObjects(_)
                     | Statement::DropOwned(_)