Ma/prom metrics/sql min

src/adapter/src/catalog.rs

@@ -1579,6 +1579,8 @@ pub(crate) fn comment_id_to_audit_object_type(id: CommentObjectId) -> ObjectType
         CommentObjectId::MaterializedView(_) => ObjectType::MaterializedView,
         CommentObjectId::Source(_) => ObjectType::Source,
         CommentObjectId::Sink(_) => ObjectType::Sink,
+        // No dedicated audit `ObjectType` for metric sinks; treat them as sinks.
+        CommentObjectId::MetricSink(_) => ObjectType::Sink,
         CommentObjectId::Index(_) => ObjectType::Index,
         CommentObjectId::Func(_) => ObjectType::Func,
         CommentObjectId::Connection(_) => ObjectType::Connection,
@@ -1609,6 +1611,8 @@ pub(crate) fn system_object_type_to_audit_object_type(
             mz_sql::catalog::ObjectType::MaterializedView => ObjectType::MaterializedView,
             mz_sql::catalog::ObjectType::Source => ObjectType::Source,
             mz_sql::catalog::ObjectType::Sink => ObjectType::Sink,
+            // No dedicated audit `ObjectType` for metric sinks; treat them as sinks.
+            mz_sql::catalog::ObjectType::MetricSink => ObjectType::Sink,
             mz_sql::catalog::ObjectType::Index => ObjectType::Index,
             mz_sql::catalog::ObjectType::Type => ObjectType::Type,
             mz_sql::catalog::ObjectType::Role => ObjectType::Role,

src/adapter/src/catalog/apply.rs

@@ -1569,6 +1569,7 @@ impl CatalogState {
         match entry.item_mut() {
             CatalogItem::Index(idx) => idx.optimized_plan = Some(Arc::new(plan)),
             CatalogItem::MaterializedView(mv) => mv.optimized_plan = Some(Arc::new(plan)),
+            CatalogItem::MetricSink(ms) => ms.optimized_plan = Some(Arc::new(plan)),
             other => panic!("set_optimized_plan called on {} ({:?})", id, other.typ()),
         }
     }
@@ -1587,6 +1588,7 @@ impl CatalogState {
         match entry.item_mut() {
             CatalogItem::Index(idx) => idx.physical_plan = Some(Arc::new(plan)),
             CatalogItem::MaterializedView(mv) => mv.physical_plan = Some(Arc::new(plan)),
+            CatalogItem::MetricSink(ms) => ms.physical_plan = Some(Arc::new(plan)),
             other => panic!("set_physical_plan called on {} ({:?})", id, other.typ()),
         }
     }
@@ -1622,6 +1624,7 @@ impl CatalogState {
         match entry.item_mut() {
             CatalogItem::Index(idx) => idx.dataflow_metainfo = Some(metainfo),
             CatalogItem::MaterializedView(mv) => mv.dataflow_metainfo = Some(metainfo),
+            CatalogItem::MetricSink(ms) => ms.dataflow_metainfo = Some(metainfo),
             other => panic!("set_dataflow_metainfo called on {} ({:?})", id, other.typ()),
         }
     }
@@ -2443,7 +2446,8 @@ 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::MetricSink => derived_items.push(update),
                 CatalogItemType::Sink => sinks.push(update),
             }
         }
@@ -2508,7 +2512,8 @@ 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::MetricSink => derived_items.push(update),
                 CatalogItemType::Sink => sinks.push(update),
             }
         }

src/adapter/src/catalog/builtin_table_updates.rs

@@ -364,7 +364,7 @@ impl CatalogState {
             CatalogItem::Func(func) => {
                 self.pack_func_update(id, schema_id, name, owner_id, func, diff)
             }
-            CatalogItem::Log(_) | CatalogItem::Secret(_) => vec![],
+            CatalogItem::Log(_) | CatalogItem::Secret(_) | CatalogItem::MetricSink(_) => vec![],
             CatalogItem::Connection(connection) => {
                 self.pack_connection_update(id, connection, diff)
             }

src/adapter/src/catalog/consistency.rs

@@ -269,6 +269,7 @@ impl CatalogState {
                 | CommentObjectId::MaterializedView(item_id)
                 | CommentObjectId::Source(item_id)
                 | CommentObjectId::Sink(item_id)
+                | CommentObjectId::MetricSink(item_id)
                 | CommentObjectId::Index(item_id)
                 | CommentObjectId::Func(item_id)
                 | CommentObjectId::Connection(item_id)

src/adapter/src/catalog/open.rs

@@ -928,6 +928,7 @@ fn add_new_remove_old_builtin_items_migration(
             CatalogItemType::View => CommentObjectId::View(id),
             CatalogItemType::MaterializedView => CommentObjectId::MaterializedView(id),
             CatalogItemType::Sink
+            | CatalogItemType::MetricSink
             | CatalogItemType::Index
             | CatalogItemType::Type
             | CatalogItemType::Func

src/adapter/src/catalog/state.rs

@@ -31,7 +31,7 @@ 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,
+    Connection, DataSourceDesc, Database, DefaultPrivileges, Index, MaterializedView, MetricSink,
     NetworkPolicy, Role, RoleAuth, Schema, Secret, Sink, Source, SourceReferences, Table,
     TableDataSource, Type, View,
 };
@@ -74,9 +74,9 @@ use mz_sql::names::{
     ResolvedDatabaseSpecifier, ResolvedIds, SchemaId, SchemaSpecifier, SystemObjectId,
 };
 use mz_sql::plan::{
-    CreateConnectionPlan, CreateIndexPlan, CreateMaterializedViewPlan, CreateSecretPlan,
-    CreateSinkPlan, CreateSourcePlan, CreateTablePlan, CreateTypePlan, CreateViewPlan, Params,
-    Plan, PlanContext,
+    CreateConnectionPlan, CreateIndexPlan, CreateMaterializedViewPlan, CreateMetricSinkPlan,
+    CreateSecretPlan, CreateSinkPlan, CreateSourcePlan, CreateTablePlan, CreateTypePlan,
+    CreateViewPlan, Params, Plan, PlanContext,
 };
 use mz_sql::rbac;
 use mz_sql::session::metadata::SessionMetadata;
@@ -458,6 +458,12 @@ impl CatalogState {
                         queue.push_back(from_item_id);
                     }
                 }
+                CatalogItem::MetricSink(metric_sink) => {
+                    let from_item_id = self.get_entry_by_global_id(&metric_sink.from).id();
+                    if seen.insert(from_item_id) {
+                        queue.push_back(from_item_id);
+                    }
+                }
                 CatalogItem::Index(idx) => {
                     let on_item_id = self.get_entry_by_global_id(&idx.on).id();
                     if seen.insert(on_item_id) {
@@ -1503,6 +1509,20 @@ impl CatalogState {
                 physical_plan: None,
                 dataflow_metainfo: None,
             }),
+            Plan::CreateMetricSink(CreateMetricSinkPlan {
+                metric_sink,
+                in_cluster,
+                ..
+            }) => CatalogItem::MetricSink(MetricSink {
+                create_sql: metric_sink.create_sql,
+                global_id,
+                from: metric_sink.from,
+                resolved_ids,
+                cluster_id: in_cluster,
+                optimized_plan: None,
+                physical_plan: None,
+                dataflow_metainfo: None,
+            }),
             Plan::CreateSink(CreateSinkPlan {
                 sink,
                 with_snapshot,
@@ -1918,6 +1938,7 @@ impl CatalogState {
             CatalogItemType::Table
             | CatalogItemType::Source
             | CatalogItemType::Sink
+            | CatalogItemType::MetricSink
             | CatalogItemType::View
             | CatalogItemType::MaterializedView
             | CatalogItemType::Index
@@ -2740,6 +2761,7 @@ impl CatalogState {
             | CommentObjectId::MaterializedView(id)
             | CommentObjectId::Source(id)
             | CommentObjectId::Sink(id)
+            | CommentObjectId::MetricSink(id)
             | CommentObjectId::Index(id)
             | CommentObjectId::Func(id)
             | CommentObjectId::Connection(id)
@@ -2769,6 +2791,7 @@ impl CatalogState {
             | CommentObjectId::MaterializedView(id)
             | CommentObjectId::Source(id)
             | CommentObjectId::Sink(id)
+            | CommentObjectId::MetricSink(id)
             | CommentObjectId::Index(id)
             | CommentObjectId::Func(id)
             | CommentObjectId::Connection(id)

src/adapter/src/catalog/timeline.rs

@@ -93,6 +93,7 @@ impl Catalog {
                         }
                         CatalogItem::View(_)
                         | CatalogItem::Sink(_)
+                        | CatalogItem::MetricSink(_)
                         | CatalogItem::Type(_)
                         | CatalogItem::Func(_)
                         | CatalogItem::Secret(_)
@@ -230,6 +231,7 @@ impl Catalog {
                         ));
                     }
                     CatalogItem::Sink(_)
+                    | CatalogItem::MetricSink(_)
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)

src/adapter/src/catalog/transact.rs

@@ -875,7 +875,8 @@ impl Catalog {
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)
-                    | CatalogItem::Connection(_) => {}
+                    | CatalogItem::Connection(_)
+                    | CatalogItem::MetricSink(_) => {}
                 }
             }
         }
@@ -1687,7 +1688,10 @@ impl Catalog {
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)
-                    | CatalogItem::Connection(_) => (),
+                    | CatalogItem::Connection(_)
+                    // A metric sink writes to the in-process metrics registry, not persist, so it
+                    // has no storage collection to create.
+                    | CatalogItem::MetricSink(_) => (),
                 }
 
                 let system_user = session.map_or(false, |s| s.user().is_system_user());
@@ -1871,7 +1875,11 @@ impl Catalog {
                         | CatalogItem::Type(_)
                         | CatalogItem::Func(_)
                         | CatalogItem::Secret(_)
-                        | CatalogItem::Connection(_) => EventDetails::IdFullNameV1(IdFullNameV1 {
+                        | CatalogItem::Connection(_)
+                        // Metric sinks reuse the generic detail and the `Sink` audit object type
+                        // (see the `From<CatalogItemType>` impl in mz_sql::catalog), so no
+                        // dedicated audit-log variant or proto bump is needed.
+                        | CatalogItem::MetricSink(_) => EventDetails::IdFullNameV1(IdFullNameV1 {
                             id: id.to_string(),
                             name,
                         }),

src/adapter/src/command.rs

@@ -611,6 +611,8 @@ pub enum ExecuteResponse {
     CreatedClusterReplica,
     /// The requested index was created.
     CreatedIndex,
+    /// The requested metric sink was created.
+    CreatedMetricSink,
     /// The requested introspection subscribe was created.
     CreatedIntrospectionSubscribe,
     /// The requested secret was created.
@@ -787,6 +789,7 @@ impl TryInto<ExecuteResponse> for ExecuteResponseKind {
                 Ok(ExecuteResponse::CreatedClusterReplica)
             }
             ExecuteResponseKind::CreatedIndex => Ok(ExecuteResponse::CreatedIndex),
+            ExecuteResponseKind::CreatedMetricSink => Ok(ExecuteResponse::CreatedMetricSink),
             ExecuteResponseKind::CreatedSecret => Ok(ExecuteResponse::CreatedSecret),
             ExecuteResponseKind::CreatedSink => Ok(ExecuteResponse::CreatedSink),
             ExecuteResponseKind::CreatedSource => Ok(ExecuteResponse::CreatedSource),
@@ -851,6 +854,7 @@ impl ExecuteResponse {
             CreatedCluster { .. } => Some("CREATE CLUSTER".into()),
             CreatedClusterReplica { .. } => Some("CREATE CLUSTER REPLICA".into()),
             CreatedIndex { .. } => Some("CREATE INDEX".into()),
+            CreatedMetricSink { .. } => Some("CREATE METRIC SINK".into()),
             CreatedSecret { .. } => Some("CREATE SECRET".into()),
             CreatedSink { .. } => Some("CREATE SINK".into()),
             CreatedSource { .. } => Some("CREATE SOURCE".into()),
@@ -950,6 +954,7 @@ impl ExecuteResponse {
             CreateView => &[CreatedView],
             CreateMaterializedView => &[CreatedMaterializedView],
             CreateIndex => &[CreatedIndex],
+            CreateMetricSink => &[CreatedMetricSink],
             CreateType => &[CreatedType],
             PlanKind::Deallocate => &[ExecuteResponseKind::Deallocate],
             CreateNetworkPolicy => &[CreatedNetworkPolicy],

src/adapter/src/coord.rs

@@ -408,6 +408,11 @@ pub enum Message {
         span: Span,
         stage: CreateIndexStage,
     },
+    CreateMetricSinkStageReady {
+        ctx: ExecuteContext,
+        span: Span,
+        stage: CreateMetricSinkStage,
+    },
     CreateViewStageReady {
         ctx: ExecuteContext,
         span: Span,
@@ -545,6 +550,7 @@ impl Message {
             Message::PeekStageReady { .. } => "peek_stage_ready",
             Message::ExplainTimestampStageReady { .. } => "explain_timestamp_stage_ready",
             Message::CreateIndexStageReady { .. } => "create_index_stage_ready",
+            Message::CreateMetricSinkStageReady { .. } => "create_metric_sink_stage_ready",
             Message::CreateViewStageReady { .. } => "create_view_stage_ready",
             Message::CreateMaterializedViewStageReady { .. } => {
                 "create_materialized_view_stage_ready"
@@ -791,6 +797,30 @@ pub struct CreateIndexExplain {
     explain_ctx: ExplainPlanContext,
 }
 
+#[derive(Debug)]
+pub enum CreateMetricSinkStage {
+    Optimize(CreateMetricSinkOptimize),
+    Finish(CreateMetricSinkFinish),
+}
+
+#[derive(Debug)]
+pub struct CreateMetricSinkOptimize {
+    validity: PlanValidity,
+    plan: plan::CreateMetricSinkPlan,
+    resolved_ids: ResolvedIds,
+}
+
+#[derive(Debug)]
+pub struct CreateMetricSinkFinish {
+    validity: PlanValidity,
+    item_id: CatalogItemId,
+    global_id: GlobalId,
+    plan: plan::CreateMetricSinkPlan,
+    resolved_ids: ResolvedIds,
+    global_mir_plan: optimize::metric_sink::GlobalMirPlan,
+    global_lir_plan: optimize::metric_sink::GlobalLirPlan,
+}
+
 #[derive(Debug)]
 pub enum CreateViewStage {
     Optimize(CreateViewOptimize),
@@ -2756,6 +2786,9 @@ impl Coordinator {
                 | CatalogItem::Type(_)
                 | CatalogItem::Func(_)
                 | CatalogItem::Secret(_) => {}
+                // Metric sinks are durable catalog items, but re-optimizing and shipping their
+                // dataflow on boot is deferred to bootstrap rehydration; skip for now.
+                CatalogItem::MetricSink(_) => {}
             }
         }
 
@@ -3341,6 +3374,9 @@ impl Coordinator {
                 | CatalogItem::Func(_)
                 | CatalogItem::Secret(_)
                 | CatalogItem::Connection(_) => (),
+                // A metric sink writes to the in-process metrics registry, not to persist, so it
+                // has no storage collection to bootstrap. Same treatment as `Index`.
+                CatalogItem::MetricSink(_) => (),
             }
         }
 
@@ -3751,6 +3787,9 @@ impl Coordinator {
                 | CatalogItem::Func(_)
                 | CatalogItem::Secret(_)
                 | CatalogItem::Connection(_) => (),
+                // No physical plan is built for a metric sink on boot; re-optimizing and shipping
+                // its dataflow is deferred to bootstrap rehydration.
+                CatalogItem::MetricSink(_) => (),
             }
         }
 
@@ -3783,6 +3822,9 @@ impl Coordinator {
                 | CatalogItem::Func(_)
                 | CatalogItem::Secret(_)
                 | CatalogItem::Connection(_) => continue,
+                // A metric sink has no persist dependency and no physical plan on boot (see
+                // `bootstrap_dataflow_plans`), so there is no as-of to select for it yet.
+                CatalogItem::MetricSink(_) => continue,
             };
             if let Some(plan) = self.catalog.try_get_physical_plan(&gid) {
                 catalog_ids.push(gid);

src/adapter/src/coord/appends.rs

@@ -1211,6 +1211,7 @@ pub(crate) fn waiting_on_startup_appends(
         | Plan::CreateView(_)
         | Plan::CreateMaterializedView(_)
         | Plan::CreateIndex(_)
+        | Plan::CreateMetricSink(_)
         | Plan::CreateType(_)
         | Plan::Comment(_)
         | Plan::DiscardTemp