diff --git a/src/adapter/src/catalog.rs b/src/adapter/src/catalog.rs
index 8b3a1f919da38..54dd06d01599d 100644
--- a/src/adapter/src/catalog.rs
+++ b/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,
diff --git a/src/adapter/src/catalog/apply.rs b/src/adapter/src/catalog/apply.rs
index cd1a23f721ae6..b4bb2f77e682c 100644
--- a/src/adapter/src/catalog/apply.rs
+++ b/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),
             }
         }
diff --git a/src/adapter/src/catalog/builtin_table_updates.rs b/src/adapter/src/catalog/builtin_table_updates.rs
index ce1c40aa83108..b8b3d5eaf3c23 100644
--- a/src/adapter/src/catalog/builtin_table_updates.rs
+++ b/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)
             }
diff --git a/src/adapter/src/catalog/consistency.rs b/src/adapter/src/catalog/consistency.rs
index 37ec77b4c107d..2a02d09477641 100644
--- a/src/adapter/src/catalog/consistency.rs
+++ b/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)
diff --git a/src/adapter/src/catalog/open.rs b/src/adapter/src/catalog/open.rs
index 693515ccca026..8fec413cf0144 100644
--- a/src/adapter/src/catalog/open.rs
+++ b/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
diff --git a/src/adapter/src/catalog/state.rs b/src/adapter/src/catalog/state.rs
index d4932ebe883be..17349aaa3cd35 100644
--- a/src/adapter/src/catalog/state.rs
+++ b/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)
diff --git a/src/adapter/src/catalog/timeline.rs b/src/adapter/src/catalog/timeline.rs
index 364fcac385bd1..f283e5d33da4e 100644
--- a/src/adapter/src/catalog/timeline.rs
+++ b/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(_)
diff --git a/src/adapter/src/catalog/transact.rs b/src/adapter/src/catalog/transact.rs
index 0d158b5eb3c4e..f13bfd3b25fda 100644
--- a/src/adapter/src/catalog/transact.rs
+++ b/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,
                         }),
diff --git a/src/adapter/src/command.rs b/src/adapter/src/command.rs
index 3174ff359e99f..05753230132ca 100644
--- a/src/adapter/src/command.rs
+++ b/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],
diff --git a/src/adapter/src/coord.rs b/src/adapter/src/coord.rs
index 29ada71ba3e6d..4bd3fa8a1c9da 100644
--- a/src/adapter/src/coord.rs
+++ b/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);
diff --git a/src/adapter/src/coord/appends.rs b/src/adapter/src/coord/appends.rs
index a0c482c179938..db2e473cfc698 100644
--- a/src/adapter/src/coord/appends.rs
+++ b/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
diff --git a/src/adapter/src/coord/catalog_implications.rs b/src/adapter/src/coord/catalog_implications.rs
index 5f3dc9b46a13d..eb8294fc69b99 100644
--- a/src/adapter/src/coord/catalog_implications.rs
+++ b/src/adapter/src/coord/catalog_implications.rs
@@ -35,7 +35,7 @@ use itertools::Itertools;
 use mz_adapter_types::compaction::CompactionWindow;
 use mz_catalog::memory::objects::{
     CatalogItem, Cluster, ClusterReplica, Connection, DataSourceDesc, Index, MaterializedView,
-    Secret, Sink, Source, StateDiff, Table, TableDataSource, View,
+    MetricSink, Secret, Sink, Source, StateDiff, Table, TableDataSource, View,
 };
 use mz_cloud_resources::VpcEndpointConfig;
 use mz_compute_client::logging::LogVariant;
@@ -237,6 +237,7 @@ impl Coordinator {
         let mut replication_slots_to_drop: Vec<(PostgresConnection, String)> = vec![];
         let mut storage_sink_gids_to_drop = vec![];
         let mut indexes_to_drop = vec![];
+        let mut metric_sinks_to_drop = vec![];
         let mut compute_sinks_to_drop = vec![];
         let mut view_gids_to_drop = vec![];
         let mut secrets_to_drop = vec![];
@@ -436,6 +437,23 @@ impl Coordinator {
                     indexes_to_drop.push((index.cluster_id, index.global_id()));
                     dropped_item_names.insert(index.global_id(), full_name);
                 }
+                CatalogImplication::MetricSink(CatalogImplicationKind::Added(_metric_sink)) => {
+                    // No controller action needed here, mirroring `Index`: create-time shipping
+                    // of the dataflow is the sequencer's job (`create_metric_sink_finish`), and
+                    // rehydration after a restart is handled separately during bootstrap
+                    // (`bootstrap_dataflow_plans`).
+                }
+                CatalogImplication::MetricSink(CatalogImplicationKind::Altered { .. }) => {
+                    // No action needed: owner, privilege, and rename changes are
+                    // catalog-only and require no controller changes.
+                }
+                CatalogImplication::MetricSink(CatalogImplicationKind::Dropped(
+                    metric_sink,
+                    full_name,
+                )) => {
+                    metric_sinks_to_drop.push((metric_sink.cluster_id, metric_sink.global_id));
+                    dropped_item_names.insert(metric_sink.global_id, full_name);
+                }
                 CatalogImplication::MaterializedView(CatalogImplicationKind::Added(mv)) => {
                     tracing::debug!(?mv, "not handling AddMaterializedView in here yet");
                 }
@@ -587,6 +605,7 @@ impl Coordinator {
                 | CatalogImplication::Source(CatalogImplicationKind::None)
                 | CatalogImplication::Sink(CatalogImplicationKind::None)
                 | CatalogImplication::Index(CatalogImplicationKind::None)
+                | CatalogImplication::MetricSink(CatalogImplicationKind::None)
                 | CatalogImplication::MaterializedView(CatalogImplicationKind::None)
                 | CatalogImplication::View(CatalogImplicationKind::None)
                 | CatalogImplication::Secret(CatalogImplicationKind::None)
@@ -804,6 +823,7 @@ impl Coordinator {
             .map(|(_, gid)| *gid)
             .chain(tables_to_drop.iter().map(|(_, gid)| *gid))
             .chain(indexes_to_drop.iter().map(|(_, gid)| *gid))
+            .chain(metric_sinks_to_drop.iter().map(|(_, gid)| *gid))
             .chain(view_gids_to_drop.iter().copied())
             .collect();
 
@@ -882,6 +902,7 @@ impl Coordinator {
             .collect();
         let compute_gids_to_drop: Vec<_> = indexes_to_drop
             .iter()
+            .chain(metric_sinks_to_drop.iter())
             .chain(compute_sinks_to_drop.iter())
             .copied()
             .collect();
@@ -1610,7 +1631,8 @@ impl Coordinator {
                     | CatalogItem::Index(_)
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
-                    | CatalogItem::Secret(_) => {
+                    | CatalogItem::Secret(_)
+                    | CatalogItem::MetricSink(_) => {
                         // Other item types don't have connection dependencies
                         // that need updating.
                     }
@@ -1671,6 +1693,7 @@ enum CatalogImplication {
     Source(CatalogImplicationKind<(Source, Option<GenericSourceConnection>)>),
     Sink(CatalogImplicationKind<Sink>),
     Index(CatalogImplicationKind<Index>),
+    MetricSink(CatalogImplicationKind<MetricSink>),
     MaterializedView(CatalogImplicationKind<MaterializedView>),
     View(CatalogImplicationKind<View>),
     Secret(CatalogImplicationKind<Secret>),
@@ -1824,6 +1847,13 @@ impl CatalogImplication {
                 CatalogItem::Connection(connection) => {
                     self.absorb_connection(connection, None, catalog_update.diff);
                 }
+                CatalogItem::MetricSink(metric_sink) => {
+                    self.absorb_metric_sink(
+                        metric_sink,
+                        Some(parsed_full_name),
+                        catalog_update.diff,
+                    );
+                }
                 CatalogItem::Log(_) => {}
                 CatalogItem::Type(_) => {}
                 CatalogItem::Func(_) => {}
@@ -1863,6 +1893,13 @@ impl CatalogImplication {
                 CatalogItem::Connection(connection) => {
                     self.absorb_connection(connection, None, catalog_update.diff);
                 }
+                CatalogItem::MetricSink(metric_sink) => {
+                    self.absorb_metric_sink(
+                        metric_sink,
+                        Some(parsed_full_name),
+                        catalog_update.diff,
+                    );
+                }
                 CatalogItem::Log(_) => {}
                 CatalogItem::Type(_) => {}
                 CatalogItem::Func(_) => {}
@@ -1907,6 +1944,7 @@ impl CatalogImplication {
     );
     impl_absorb_method!(absorb_sink, Sink, Sink);
     impl_absorb_method!(absorb_index, Index, Index);
+    impl_absorb_method!(absorb_metric_sink, MetricSink, MetricSink);
     impl_absorb_method!(absorb_materialized_view, MaterializedView, MaterializedView);
     impl_absorb_method!(absorb_view, View, View);
 
diff --git a/src/adapter/src/coord/catalog_serving.rs b/src/adapter/src/coord/catalog_serving.rs
index 20a3a40372934..90f2be772938f 100644
--- a/src/adapter/src/coord/catalog_serving.rs
+++ b/src/adapter/src/coord/catalog_serving.rs
@@ -91,6 +91,7 @@ pub fn auto_run_on_catalog_server<'a, 's, 'p>(
         | Plan::CreateView(_)
         | Plan::CreateMaterializedView(_)
         | Plan::CreateIndex(_)
+        | Plan::CreateMetricSink(_)
         | Plan::CreateType(_)
         | Plan::Comment(_)
         | Plan::DiscardTemp
diff --git a/src/adapter/src/coord/command_handler.rs b/src/adapter/src/coord/command_handler.rs
index e6046a8a501e9..0eb49745ed06e 100644
--- a/src/adapter/src/coord/command_handler.rs
+++ b/src/adapter/src/coord/command_handler.rs
@@ -1379,6 +1379,7 @@ impl Coordinator {
                     | Statement::CreateSchema(_)
                     | Statement::CreateSecret(_)
                     | Statement::CreateSink(_)
+                    | Statement::CreateMetricSink(_)
                     | Statement::CreateSubsource(_)
                     | Statement::CreateTable(_)
                     | Statement::CreateType(_)
diff --git a/src/adapter/src/coord/ddl.rs b/src/adapter/src/coord/ddl.rs
index 3aa7420598830..bb33587b06de6 100644
--- a/src/adapter/src/coord/ddl.rs
+++ b/src/adapter/src/coord/ddl.rs
@@ -342,10 +342,6 @@ impl Coordinator {
                 catalog::Op::DropObjects(drop_object_infos) => {
                     for drop_object_info in drop_object_infos {
                         match &drop_object_info {
-                            catalog::DropObjectInfo::Item(_) => {
-                                // Nothing to do, these will be handled by
-                                // applying the side effects that we return.
-                            }
                             catalog::DropObjectInfo::Cluster(id) => {
                                 clusters_to_drop.push(*id);
                             }
@@ -1211,7 +1207,8 @@ impl Coordinator {
                         | CatalogItem::View(_)
                         | CatalogItem::Index(_)
                         | CatalogItem::Type(_)
-                        | CatalogItem::Func(_) => {}
+                        | CatalogItem::Func(_)
+                        | CatalogItem::MetricSink(_) => {}
                     }
                 }
                 Op::DropObjects(drop_object_infos) => {
@@ -1290,7 +1287,8 @@ impl Coordinator {
                                     | CatalogItem::View(_)
                                     | CatalogItem::Index(_)
                                     | CatalogItem::Type(_)
-                                    | CatalogItem::Func(_) => {}
+                                    | CatalogItem::Func(_)
+                                    | CatalogItem::MetricSink(_) => {}
                                 }
                             }
                         }
@@ -1320,7 +1318,8 @@ impl Coordinator {
                     | CatalogItem::View(_)
                     | CatalogItem::Index(_)
                     | CatalogItem::Type(_)
-                    | CatalogItem::Func(_) => {}
+                    | CatalogItem::Func(_)
+                    | CatalogItem::MetricSink(_) => {}
                 },
                 Op::AlterRole { .. }
                 | Op::AlterRetainHistory { .. }
diff --git a/src/adapter/src/coord/indexes.rs b/src/adapter/src/coord/indexes.rs
index 1328ba83129d3..8d1cfc78c1e97 100644
--- a/src/adapter/src/coord/indexes.rs
+++ b/src/adapter/src/coord/indexes.rs
@@ -69,6 +69,7 @@ impl DataflowBuilder<'_> {
                         panic!("log source {id} is missing index");
                     }
                     CatalogItem::Sink(_)
+                    | CatalogItem::MetricSink(_)
                     | CatalogItem::Index(_)
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
diff --git a/src/adapter/src/coord/message_handler.rs b/src/adapter/src/coord/message_handler.rs
index c9eb4e0e17c95..929d7b9239b28 100644
--- a/src/adapter/src/coord/message_handler.rs
+++ b/src/adapter/src/coord/message_handler.rs
@@ -204,6 +204,9 @@ impl Coordinator {
             Message::CreateIndexStageReady { ctx, span, stage } => {
                 self.sequence_staged(ctx, span, stage).boxed_local().await;
             }
+            Message::CreateMetricSinkStageReady { ctx, span, stage } => {
+                self.sequence_staged(ctx, span, stage).boxed_local().await;
+            }
             Message::CreateViewStageReady { ctx, span, stage } => {
                 self.sequence_staged(ctx, span, stage).boxed_local().await;
             }
diff --git a/src/adapter/src/coord/read_then_write.rs b/src/adapter/src/coord/read_then_write.rs
index ff2bdd336d69b..67e0c50d1ce5c 100644
--- a/src/adapter/src/coord/read_then_write.rs
+++ b/src/adapter/src/coord/read_then_write.rs
@@ -98,7 +98,7 @@ pub(crate) fn validate_read_then_write_dependencies(
                     }
                     Source | Secret | Connection => false,
                     // Cannot select from sinks or indexes.
-                    Sink | Index => unreachable!(),
+                    Sink | MetricSink | Index => unreachable!(),
                     Table => {
                         if !id.is_user() {
                             // We can't read from non-user tables
diff --git a/src/adapter/src/coord/sequencer.rs b/src/adapter/src/coord/sequencer.rs
index d5229268c7de7..3b174dbf03826 100644
--- a/src/adapter/src/coord/sequencer.rs
+++ b/src/adapter/src/coord/sequencer.rs
@@ -292,6 +292,10 @@ impl Coordinator {
                 Plan::CreateIndex(plan) => {
                     self.sequence_create_index(ctx, plan, resolved_ids).await;
                 }
+                Plan::CreateMetricSink(plan) => {
+                    self.sequence_create_metric_sink(ctx, plan, resolved_ids)
+                        .await;
+                }
                 Plan::CreateType(plan) => {
                     let result = self
                         .sequence_create_type(ctx.session(), plan, resolved_ids)
diff --git a/src/adapter/src/coord/sequencer/inner.rs b/src/adapter/src/coord/sequencer/inner.rs
index f979d56383ef1..028e53f58a5c6 100644
--- a/src/adapter/src/coord/sequencer/inner.rs
+++ b/src/adapter/src/coord/sequencer/inner.rs
@@ -18,6 +18,7 @@ use anyhow::anyhow;
 use futures::future::{BoxFuture, FutureExt};
 use futures::{Future, StreamExt, future};
 use itertools::Itertools;
+use maplit::btreemap;
 use mz_adapter_types::compaction::CompactionWindow;
 use mz_adapter_types::connection::ConnectionId;
 use mz_adapter_types::dyncfgs::{ENABLE_PASSWORD_AUTH, READ_THEN_WRITE_MAX_DEPENDENCIES};
@@ -25,6 +26,9 @@ use mz_catalog::memory::error::ErrorKind;
 use mz_catalog::memory::objects::{
     CatalogItem, Connection, DataSourceDesc, Sink, Source, Table, TableDataSource, Type,
 };
+use mz_compute_types::ComputeInstanceId;
+use mz_compute_types::dataflows::DataflowDescription;
+use mz_compute_types::plan::LirRelationExpr;
 use mz_expr::{
     CollectionPlan, Eval, MapFilterProject, OptimizedMirRelationExpr, ResultSpec, RowSetFinishing,
 };
@@ -38,10 +42,11 @@ use mz_repr::adt::jsonb::Jsonb;
 use mz_repr::adt::mz_acl_item::{MzAclItem, PrivilegeMap};
 use mz_repr::explain::ExprHumanizer;
 use mz_repr::explain::json::json_string;
+use mz_repr::explain::{ExprHumanizerExt, TransientItem};
 use mz_repr::role_id::RoleId;
 use mz_repr::{
-    CatalogItemId, Datum, Diff, GlobalId, RelationVersion, RelationVersionSelector, Row, RowArena,
-    RowIterator, Timestamp,
+    CatalogItemId, Datum, Diff, GlobalId, RelationDesc, RelationVersion, RelationVersionSelector,
+    Row, RowArena, RowIterator, Timestamp,
 };
 use mz_secrets::SecretsReader;
 use mz_sql::ast::{
@@ -97,7 +102,9 @@ use timely::progress::Antichain;
 use tokio::sync::{oneshot, watch};
 use tracing::{Instrument, Span, info, warn};
 
-use crate::catalog::{self, Catalog, ConnCatalog, DropObjectInfo, UpdatePrivilegeVariant};
+use crate::catalog::{
+    self, Catalog, CatalogState, ConnCatalog, DropObjectInfo, UpdatePrivilegeVariant,
+};
 use crate::command::{ExecuteResponse, Response};
 use crate::coord::appends::{
     BuiltinTableAppendNotify, DeferredOp, DeferredPlan, PendingWriteTxn, UserWriteResponder,
@@ -120,7 +127,7 @@ use crate::session::{
     WriteLocks, WriteOp,
 };
 use crate::util::{ClientTransmitter, ResultExt, viewable_variables};
-use crate::{PeekResponseUnary, ReadHolds};
+use crate::{CollectionIdBundle, PeekResponseUnary, ReadHolds};
 
 /// A future that resolves to a real-time recency timestamp.
 type RtrTimestampFuture = BoxFuture<'static, Result<Timestamp, StorageError>>;
@@ -129,6 +136,7 @@ mod cluster;
 mod copy_from;
 mod create_index;
 mod create_materialized_view;
+mod create_metric_sink;
 mod create_view;
 mod explain_timestamp;
 mod peek;
@@ -4944,6 +4952,71 @@ impl Coordinator {
         emit_optimizer_notices(&*self.catalog, ctx.session(), notices);
     }
 
+    /// Renders `raw_df_meta`'s optimizer notices against a humanizer that resolves the
+    /// about-to-be-created item's own `global_id` to `name`, rather than to a bare transient
+    /// id. `source_desc` supplies the target's column names for humanized output.
+    ///
+    /// The `CREATE METRIC SINK` `_finish` stage renders notices before the catalog transaction
+    /// that creates the item, so the rendered text can already refer to the item by its
+    /// intended name.
+    fn render_create_item_notices(
+        &self,
+        name: &QualifiedItemName,
+        global_id: GlobalId,
+        source_desc: &RelationDesc,
+        raw_df_meta: &DataflowMetainfo,
+    ) -> DataflowMetainfo<Arc<OptimizerNotice>> {
+        let notice_ids = std::iter::repeat_with(|| self.allocate_transient_id())
+            .map(|(_item_id, notice_id)| notice_id)
+            .take(raw_df_meta.optimizer_notices.len())
+            .collect::<Vec<_>>();
+
+        let system_catalog = self.catalog().for_system_session();
+        let full_name = self.catalog().resolve_full_name(name, None);
+        let transient_items = btreemap! {
+            global_id => TransientItem::new(
+                Some(full_name.into_parts()),
+                Some(source_desc.iter_names().map(|c| c.to_string()).collect()),
+            )
+        };
+        let humanizer = ExprHumanizerExt::new(transient_items, &system_catalog);
+        CatalogState::render_notices_core(
+            &humanizer,
+            (self.catalog().config().now)(),
+            raw_df_meta,
+            notice_ids,
+            Some(global_id),
+        )
+    }
+
+    /// Sets `df_desc`'s as-of from a read hold on `id_bundle`, ships the dataflow, and drops the
+    /// hold once compute has taken its own (compute puts in its own read holds during
+    /// `create_dataflow`, so it is safe to release this one right after shipping).
+    ///
+    /// The read hold across shipping keeps the since of `id_bundle` from advancing underneath the
+    /// as-of just picked.
+    async fn ship_new_dataflow(
+        &mut self,
+        id_bundle: &CollectionIdBundle,
+        mut df_desc: DataflowDescription<LirRelationExpr>,
+        instance: ComputeInstanceId,
+        notice_builtin_updates_fut: Option<BuiltinTableAppendNotify>,
+    ) {
+        let read_holds = self.acquire_read_holds(id_bundle);
+        let since = read_holds.least_valid_read();
+        df_desc.set_as_of(since);
+
+        self.ship_dataflow_and_notice_builtin_table_updates(
+            df_desc,
+            instance,
+            notice_builtin_updates_fut,
+            None,
+        )
+        .await;
+
+        drop(read_holds);
+    }
+
     /// Persist already-rendered optimizer notices for a newly created
     /// non-transient dataflow.
     ///
diff --git a/src/adapter/src/coord/sequencer/inner/create_metric_sink.rs b/src/adapter/src/coord/sequencer/inner/create_metric_sink.rs
new file mode 100644
index 0000000000000..d83799a0b795a
--- /dev/null
+++ b/src/adapter/src/coord/sequencer/inner/create_metric_sink.rs
@@ -0,0 +1,272 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+use mz_catalog::memory::error::ErrorKind;
+use mz_catalog::memory::objects::{CatalogItem, MetricSink};
+use mz_ore::instrument;
+use mz_repr::optimize::OverrideFrom;
+use mz_sql::catalog::CatalogError;
+use mz_sql::names::ResolvedIds;
+use mz_sql::plan;
+use mz_sql::session::metadata::SessionMetadata;
+use tracing::Span;
+
+use crate::command::ExecuteResponse;
+use crate::coord::sequencer::inner::return_if_err;
+use crate::coord::{
+    Coordinator, CreateMetricSinkFinish, CreateMetricSinkOptimize, CreateMetricSinkStage, Message,
+    PlanValidity, StageResult, Staged,
+};
+use crate::error::AdapterError;
+use crate::optimize::dataflows::dataflow_import_id_bundle;
+use crate::optimize::{self, Optimize};
+use crate::session::Session;
+use crate::{AdapterNotice, ExecuteContext, catalog};
+
+impl Staged for CreateMetricSinkStage {
+    type Ctx = ExecuteContext;
+
+    fn validity(&mut self) -> &mut PlanValidity {
+        match self {
+            Self::Optimize(stage) => &mut stage.validity,
+            Self::Finish(stage) => &mut stage.validity,
+        }
+    }
+
+    async fn stage(
+        self,
+        coord: &mut Coordinator,
+        ctx: &mut ExecuteContext,
+    ) -> Result<StageResult<Box<Self>>, AdapterError> {
+        match self {
+            CreateMetricSinkStage::Optimize(stage) => {
+                coord.create_metric_sink_optimize(stage).await
+            }
+            CreateMetricSinkStage::Finish(stage) => {
+                coord.create_metric_sink_finish(ctx, stage).await
+            }
+        }
+    }
+
+    fn message(self, ctx: ExecuteContext, span: Span) -> Message {
+        Message::CreateMetricSinkStageReady {
+            ctx,
+            span,
+            stage: self,
+        }
+    }
+
+    fn cancel_enabled(&self) -> bool {
+        true
+    }
+}
+
+impl Coordinator {
+    #[instrument]
+    pub(crate) async fn sequence_create_metric_sink(
+        &mut self,
+        ctx: ExecuteContext,
+        plan: plan::CreateMetricSinkPlan,
+        resolved_ids: ResolvedIds,
+    ) {
+        let stage = return_if_err!(
+            self.create_metric_sink_validate(ctx.session(), plan, resolved_ids),
+            ctx
+        );
+        self.sequence_staged(ctx, Span::current(), stage).await;
+    }
+
+    #[instrument]
+    fn create_metric_sink_validate(
+        &self,
+        session: &Session,
+        plan: plan::CreateMetricSinkPlan,
+        resolved_ids: ResolvedIds,
+    ) -> Result<CreateMetricSinkStage, AdapterError> {
+        // Track the target cluster and resolved dependencies so concurrent drops are caught
+        // between stages instead of panicking later when the dataflow is shipped.
+        let validity = PlanValidity::new(
+            self.catalog(),
+            resolved_ids.items().copied().collect(),
+            Some(plan.in_cluster),
+            None,
+            session.role_metadata().clone(),
+        );
+        Ok(CreateMetricSinkStage::Optimize(CreateMetricSinkOptimize {
+            validity,
+            plan,
+            resolved_ids,
+        }))
+    }
+
+    #[instrument]
+    async fn create_metric_sink_optimize(
+        &mut self,
+        CreateMetricSinkOptimize {
+            validity,
+            plan,
+            resolved_ids,
+        }: CreateMetricSinkOptimize,
+    ) -> Result<StageResult<Box<CreateMetricSinkStage>>, AdapterError> {
+        let cluster_id = plan.in_cluster;
+
+        // Collect optimizer parameters.
+        let compute_instance = self
+            .instance_snapshot(cluster_id)
+            .expect("compute instance does not exist");
+        let (item_id, global_id) = self.allocate_user_id().await?;
+        // A transient id for the view the optimizer builds over `from` to shape its rows (see
+        // `optimize::metric_sink::shape_metric_sink_source`); scoped to this dataflow, not durable.
+        let (_, view_id) = self.allocate_transient_id();
+
+        let optimizer_config = optimize::OptimizerConfig::from(self.catalog().system_config())
+            .override_from(&self.catalog.get_cluster(cluster_id).config.features())
+            .override_from(&self.cluster_scoped_optimizer_overrides(cluster_id));
+
+        // Build an optimizer for this METRIC SINK.
+        let mut optimizer = optimize::metric_sink::Optimizer::new(
+            self.owned_catalog(),
+            compute_instance,
+            view_id,
+            global_id,
+            optimizer_config,
+            self.optimizer_metrics(),
+        );
+        let span = Span::current();
+        Ok(StageResult::Handle(mz_ore::task::spawn_blocking(
+            || "optimize create metric sink",
+            move || {
+                span.in_scope(|| {
+                    let metric_sink = optimize::metric_sink::MetricSink::new(
+                        plan.name.clone(),
+                        plan.metric_sink.from,
+                    );
+
+                    // MIR ⇒ MIR optimization (global)
+                    let global_mir_plan = optimizer.catch_unwind_optimize(metric_sink)?;
+                    // MIR ⇒ LIR lowering and LIR ⇒ LIR optimization (global)
+                    let global_lir_plan =
+                        optimizer.catch_unwind_optimize(global_mir_plan.clone())?;
+
+                    let stage = CreateMetricSinkStage::Finish(CreateMetricSinkFinish {
+                        validity,
+                        item_id,
+                        global_id,
+                        plan,
+                        resolved_ids,
+                        global_mir_plan,
+                        global_lir_plan,
+                    });
+                    Ok(Box::new(stage))
+                })
+            },
+        )))
+    }
+
+    #[instrument]
+    async fn create_metric_sink_finish(
+        &mut self,
+        ctx: &mut ExecuteContext,
+        stage: CreateMetricSinkFinish,
+    ) -> Result<StageResult<Box<CreateMetricSinkStage>>, AdapterError> {
+        let CreateMetricSinkFinish {
+            item_id,
+            global_id,
+            plan:
+                plan::CreateMetricSinkPlan {
+                    name,
+                    metric_sink,
+                    if_not_exists,
+                    in_cluster,
+                },
+            resolved_ids,
+            global_mir_plan,
+            global_lir_plan,
+            ..
+        } = stage;
+        let id_bundle = dataflow_import_id_bundle(global_lir_plan.df_desc(), in_cluster);
+
+        let owner_id = *ctx.session().current_role_id();
+        let ops = vec![catalog::Op::CreateItem {
+            id: item_id,
+            name: name.clone(),
+            item: CatalogItem::MetricSink(MetricSink {
+                create_sql: metric_sink.create_sql,
+                global_id,
+                from: metric_sink.from,
+                resolved_ids,
+                cluster_id: metric_sink.cluster_id,
+                optimized_plan: None,
+                physical_plan: None,
+                dataflow_metainfo: None,
+            }),
+            owner_id,
+        }];
+
+        // Render optimizer notices before the catalog transaction: this way notice text resolves
+        // the new sink's own `global_id` to its intended human-readable name rather than a bare
+        // transient id.
+        let (df_desc, raw_df_meta) = global_lir_plan.unapply();
+        let from_entry = self.catalog().get_entry_by_global_id(&metric_sink.from);
+        let from_desc = from_entry
+            .relation_desc()
+            .expect("can only create a metric sink on items with a valid description");
+        let df_meta = self.render_create_item_notices(&name, global_id, &from_desc, &raw_df_meta);
+
+        let transact_result = self
+            .catalog_transact_with_side_effects(Some(ctx), ops, move |coord, _ctx| {
+                Box::pin(async move {
+                    // Save plan structures.
+                    coord
+                        .catalog_mut()
+                        .set_optimized_plan(global_id, global_mir_plan.df_desc().clone());
+                    coord
+                        .catalog_mut()
+                        .set_physical_plan(global_id, df_desc.clone());
+
+                    let notice_builtin_updates_fut =
+                        coord.persist_dataflow_metainfo(df_meta, global_id);
+
+                    // `ship_new_dataflow` puts in place a read hold across shipping, so that
+                    // update_read_capabilities can successfully act on `id_bundle`: otherwise the
+                    // since of dependencies might move along concurrently, pulling the rug from
+                    // under us.
+                    coord
+                        .ship_new_dataflow(
+                            &id_bundle,
+                            df_desc,
+                            in_cluster,
+                            notice_builtin_updates_fut,
+                        )
+                        .await;
+                    // No `allow_writes` here: metric sinks write to the in-process metrics
+                    // registry, not to external/persist state.
+                })
+            })
+            .await;
+
+        match transact_result {
+            Ok(_) => {
+                self.emit_raw_optimizer_notices_to_user(ctx, &raw_df_meta.optimizer_notices);
+                Ok(StageResult::Response(ExecuteResponse::CreatedMetricSink))
+            }
+            Err(AdapterError::Catalog(mz_catalog::memory::error::Error {
+                kind: ErrorKind::Sql(CatalogError::ItemAlreadyExists(_, _)),
+            })) if if_not_exists => {
+                ctx.session()
+                    .add_notice(AdapterNotice::ObjectAlreadyExists {
+                        name: name.item,
+                        ty: "metric sink",
+                    });
+                Ok(StageResult::Response(ExecuteResponse::CreatedMetricSink))
+            }
+            Err(err) => Err(err),
+        }
+    }
+}
diff --git a/src/adapter/src/optimize.rs b/src/adapter/src/optimize.rs
index 1c842d06c0336..97dc4bd19c48c 100644
--- a/src/adapter/src/optimize.rs
+++ b/src/adapter/src/optimize.rs
@@ -56,7 +56,7 @@ pub mod copy_to;
 pub mod dataflows;
 pub mod index;
 pub mod materialized_view;
-mod metric_sink;
+pub mod metric_sink;
 pub mod peek;
 pub mod subscribe;
 pub mod view;
diff --git a/src/adapter/src/optimize/dataflows.rs b/src/adapter/src/optimize/dataflows.rs
index eb073c1682eab..f06c50362e517 100644
--- a/src/adapter/src/optimize/dataflows.rs
+++ b/src/adapter/src/optimize/dataflows.rs
@@ -368,6 +368,7 @@ impl<'a> DataflowBuilder<'a> {
                         dataflow.import_source(*id, log.variant.desc().typ().clone(), monotonic);
                     }
                     CatalogItem::Sink(_)
+                    | CatalogItem::MetricSink(_)
                     | CatalogItem::Index(_)
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
@@ -552,6 +553,7 @@ impl<'a> DataflowBuilder<'a> {
                 | CatalogItem::Log(_)
                 | CatalogItem::MaterializedView(_)
                 | CatalogItem::Sink(_)
+                | CatalogItem::MetricSink(_)
                 | CatalogItem::Func(_) => Ok(false),
             }
         })?;
diff --git a/src/adapter/src/optimize/metric_sink.rs b/src/adapter/src/optimize/metric_sink.rs
index d0bc8eadd8619..17dbcad122ec2 100644
--- a/src/adapter/src/optimize/metric_sink.rs
+++ b/src/adapter/src/optimize/metric_sink.rs
@@ -18,23 +18,250 @@
 //! family-conflict counting) stays in the operator, because it needs the frontier-gated fold that
 //! a per-row `Map` in MIR can't express.
 
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use mz_compute_types::plan::LirRelationExpr;
+use mz_compute_types::sinks::{ComputeSinkConnection, ComputeSinkDesc, MetricSinkConnection};
 use mz_expr::func::variadic::Coalesce;
 use mz_expr::{MirRelationExpr, MirScalarExpr, func};
+use mz_repr::explain::trace_plan;
 use mz_repr::{
     ColumnName, Datum, GlobalId, RelationDesc, ReprRelationType, ReprScalarType, Row, SqlScalarType,
 };
+use mz_sql::names::QualifiedItemName;
+use mz_sql::optimizer_metrics::OptimizerMetrics;
+use mz_transform::TransformCtx;
+use mz_transform::dataflow::DataflowMetainfo;
+use mz_transform::normalize_lets::normalize_lets;
+use mz_transform::typecheck::{SharedTypecheckingContext, empty_typechecking_context};
+use timely::progress::Antichain;
+
+use crate::optimize::dataflows::{
+    ComputeInstanceSnapshot, DataflowBuilder, ExprPrep, ExprPrepMaintained,
+};
+use crate::optimize::{
+    LirDataflowDescription, MirDataflowDescription, Optimize, OptimizerCatalog, OptimizerConfig,
+    OptimizerError, optimize_mir_local,
+};
 
 /// Matches Prometheus's metric name grammar: `[a-zA-Z_:][a-zA-Z0-9_:]*`.
 ///
 /// Expressed in MIR (see `shape_metric_sink_source`) rather than parsed from a `&str` on the
 /// operator's hot path.
-// NOTE: `shape_metric_sink_source` (and this pattern) have no caller yet, hence the
-// `#[allow(dead_code)]`, but adding them alongside the compute operator
-// (`mz_compute::sink::metric_sink`) that reads the `metric_kind`/`name_valid`
-// column contract, so they live together.
-#[allow(dead_code)]
 const METRIC_NAME_PATTERN: &str = "^[a-zA-Z_:][a-zA-Z0-9_:]*$";
 
+/// Optimizer for `CREATE METRIC SINK`.
+///
+/// Like `CREATE INDEX`, the pipeline starts directly from the `GlobalId` of the collection to
+/// export rather than lowering a new relational expression from HIR. Unlike a materialized view
+/// sink, there is no persist shard, so there is no storage-metadata stage.
+pub struct Optimizer {
+    /// A representation typechecking context to use throughout the optimizer pipeline.
+    typecheck_ctx: SharedTypecheckingContext,
+    /// A snapshot of the catalog state.
+    catalog: Arc<dyn OptimizerCatalog>,
+    /// A snapshot of the cluster that will run the dataflow.
+    compute_instance: ComputeInstanceSnapshot,
+    /// A transient GlobalId for the shaped view built over the sink's source relation (see
+    /// `shape_metric_sink_source`).
+    view_id: GlobalId,
+    /// A durable GlobalId to be used with the exported metric sink.
+    sink_id: GlobalId,
+    /// Optimizer config.
+    config: OptimizerConfig,
+    /// Optimizer metrics.
+    metrics: OptimizerMetrics,
+    /// The time spent performing optimization so far.
+    duration: Duration,
+}
+
+impl Optimizer {
+    pub fn new(
+        catalog: Arc<dyn OptimizerCatalog>,
+        compute_instance: ComputeInstanceSnapshot,
+        view_id: GlobalId,
+        sink_id: GlobalId,
+        config: OptimizerConfig,
+        metrics: OptimizerMetrics,
+    ) -> Self {
+        Self {
+            typecheck_ctx: empty_typechecking_context(),
+            catalog,
+            compute_instance,
+            view_id,
+            sink_id,
+            config,
+            metrics,
+            duration: Default::default(),
+        }
+    }
+}
+
+/// A wrapper of metric sink parts needed to start the optimization process.
+pub struct MetricSink {
+    name: QualifiedItemName,
+    from: GlobalId,
+}
+
+impl MetricSink {
+    /// Construct a new [`MetricSink`]. Arguments are recorded as-is.
+    pub fn new(name: QualifiedItemName, from: GlobalId) -> Self {
+        Self { name, from }
+    }
+}
+
+/// The (sealed intermediate) result after embedding a [`MetricSink`] into a
+/// [`MirDataflowDescription`], inlining referenced views, and jointly optimizing the `MIR` plans.
+#[derive(Clone, Debug)]
+pub struct GlobalMirPlan {
+    df_desc: MirDataflowDescription,
+    df_meta: DataflowMetainfo,
+}
+
+impl GlobalMirPlan {
+    pub fn df_desc(&self) -> &MirDataflowDescription {
+        &self.df_desc
+    }
+}
+
+/// The (final) result after MIR ⇒ LIR lowering and optimizing the resulting
+/// `DataflowDescription` with `LIR` plans.
+#[derive(Clone, Debug)]
+pub struct GlobalLirPlan {
+    df_desc: LirDataflowDescription,
+    df_meta: DataflowMetainfo,
+}
+
+impl GlobalLirPlan {
+    pub fn df_desc(&self) -> &LirDataflowDescription {
+        &self.df_desc
+    }
+}
+
+impl Optimize<MetricSink> for Optimizer {
+    type To = GlobalMirPlan;
+
+    fn optimize(&mut self, metric_sink: MetricSink) -> Result<Self::To, OptimizerError> {
+        let time = Instant::now();
+
+        let from_entry = self.catalog.get_entry(&metric_sink.from);
+        let full_name = self
+            .catalog
+            .resolve_full_name(&metric_sink.name, from_entry.conn_id());
+        let from_desc = from_entry
+            .relation_desc()
+            .expect("can only create a metric sink on items with a valid description")
+            .into_owned();
+
+        let mut df_builder = {
+            let compute = self.compute_instance.clone();
+            DataflowBuilder::new(&*self.catalog, compute).with_config(&self.config)
+        };
+        let mut df_desc = MirDataflowDescription::new(full_name.to_string());
+        let mut df_meta = DataflowMetainfo::default();
+
+        df_builder.import_into_dataflow(&metric_sink.from, &mut df_desc, &self.config.features)?;
+        df_builder.maybe_reoptimize_imported_views(&mut df_desc, &self.config)?;
+
+        // Push the pure row-wise shaping (coalesce identity elements, classify the metric kind,
+        // validate the metric name) into MIR, so the operator only does the cross-row logic
+        // (dedup/collision/family-conflict) that needs the fold. See `shape_metric_sink_source`.
+        let (shaped_expr, shaped_desc) = shape_metric_sink_source(metric_sink.from, &from_desc);
+        let mut local_ctx = TransformCtx::local(
+            &self.config.features,
+            &self.typecheck_ctx,
+            &mut df_meta,
+            Some(&mut self.metrics),
+            Some(self.view_id),
+        );
+        let shaped_expr = optimize_mir_local(shaped_expr, &mut local_ctx)?;
+
+        df_builder.import_view_into_dataflow(
+            &self.view_id,
+            &shaped_expr,
+            &mut df_desc,
+            &self.config.features,
+        )?;
+        df_builder.maybe_reoptimize_imported_views(&mut df_desc, &self.config)?;
+
+        let sink_description = ComputeSinkDesc {
+            from: self.view_id,
+            from_desc: shaped_desc,
+            connection: ComputeSinkConnection::MetricSink(MetricSinkConnection {}),
+            with_snapshot: true,
+            up_to: Antichain::new(),
+            non_null_assertions: Vec::new(),
+            refresh_schedule: None,
+        };
+        df_desc.export_sink(self.sink_id, sink_description);
+
+        // Prepare expressions in the assembled dataflow.
+        let style = ExprPrepMaintained;
+        df_desc.visit_children(
+            |r| style.prep_relation_expr(r),
+            |s| style.prep_scalar_expr(s),
+        )?;
+
+        // Construct TransformCtx for global optimization.
+        let mut transform_ctx = TransformCtx::global(
+            &df_builder,
+            &mz_transform::EmptyStatisticsOracle,
+            &self.config.features,
+            &self.typecheck_ctx,
+            &mut df_meta,
+            Some(&mut self.metrics),
+        );
+        // Run global optimization.
+        mz_transform::optimize_dataflow(&mut df_desc, &mut transform_ctx, false)?;
+
+        self.duration += time.elapsed();
+
+        Ok(GlobalMirPlan { df_desc, df_meta })
+    }
+}
+
+impl Optimize<GlobalMirPlan> for Optimizer {
+    type To = GlobalLirPlan;
+
+    fn optimize(&mut self, plan: GlobalMirPlan) -> Result<Self::To, OptimizerError> {
+        let time = Instant::now();
+
+        let GlobalMirPlan {
+            mut df_desc,
+            df_meta,
+        } = plan;
+
+        // Ensure all expressions are normalized before finalizing.
+        for build in df_desc.objects_to_build.iter_mut() {
+            normalize_lets(&mut build.plan.0, &self.config.features)?
+        }
+
+        // Finalize the dataflow: MIR ⇒ LIR lowering and LIR ⇒ LIR transforms.
+        let df_desc = LirRelationExpr::finalize_dataflow(
+            df_desc,
+            &self.config.features,
+            Some(self.metrics.lowering()),
+        )?;
+
+        // Trace the pipeline output under `optimize`.
+        trace_plan(&df_desc);
+
+        self.duration += time.elapsed();
+        self.metrics
+            .observe_e2e_optimization_time("metric_sink", self.duration);
+
+        Ok(GlobalLirPlan { df_desc, df_meta })
+    }
+}
+
+impl GlobalLirPlan {
+    /// Unwraps the parts of the final result of the optimization pipeline.
+    pub fn unapply(self) -> (LirDataflowDescription, DataflowMetainfo) {
+        (self.df_desc, self.df_meta)
+    }
+}
+
 /// Extends the metric sink's imported relation with the row-wise shaping the operator otherwise
 /// has to do in Rust: coalesces `labels`/`help` to their identity element, and adds two columns
 /// the operator reads instead of parsing strings on its hot path:
@@ -53,7 +280,6 @@ const METRIC_NAME_PATTERN: &str = "^[a-zA-Z_:][a-zA-Z0-9_:]*$";
 /// via `Reduce` + `FirstValue`), collapsing the operator to a plain fold over the live set. That
 /// full move is deferred: the tiebreak fidelity that logic needs is easier to keep correct
 /// hand-written and unit-tested for now.
-#[allow(dead_code)]
 fn shape_metric_sink_source(
     from_id: GlobalId,
     from_desc: &RelationDesc,
diff --git a/src/adapter/src/statement_logging.rs b/src/adapter/src/statement_logging.rs
index 1641e545ea371..602c002b53b19 100644
--- a/src/adapter/src/statement_logging.rs
+++ b/src/adapter/src/statement_logging.rs
@@ -230,6 +230,7 @@ impl From<&ExecuteResponse> for StatementEndedExecutionReason {
             | ExecuteResponse::CreatedClusterReplica
             | ExecuteResponse::CreatedIndex
             | ExecuteResponse::CreatedIntrospectionSubscribe
+            | ExecuteResponse::CreatedMetricSink
             | ExecuteResponse::CreatedSecret
             | ExecuteResponse::CreatedSink
             | ExecuteResponse::CreatedSource
diff --git a/src/catalog-protos/src/serialization.rs b/src/catalog-protos/src/serialization.rs
index 14379bdebf1e2..1c25ae8f70ae4 100644
--- a/src/catalog-protos/src/serialization.rs
+++ b/src/catalog-protos/src/serialization.rs
@@ -223,6 +223,14 @@ impl RustType<crate::objects::CatalogItemType> for CatalogItemType {
             CatalogItemType::Func => crate::objects::CatalogItemType::Func,
             CatalogItemType::Secret => crate::objects::CatalogItemType::Secret,
             CatalogItemType::Connection => crate::objects::CatalogItemType::Connection,
+            // This conversion is only exercised for `GidMappingKey`/`SystemObjectDescription`,
+            // which describe builtin (system) objects and are populated from
+            // `Builtin::catalog_item_type`. Metric sinks are created only as user items via
+            // `CREATE METRIC SINK`, there is no `Builtin::MetricSink` variant, so this arm is
+            // never reached.
+            CatalogItemType::MetricSink => {
+                unreachable!("metric sinks are never builtin/system objects")
+            }
         }
     }
 
@@ -265,6 +273,13 @@ impl RustType<crate::objects::ObjectType> for ObjectType {
             ObjectType::Schema => crate::objects::ObjectType::Schema,
             ObjectType::Func => crate::objects::ObjectType::Func,
             ObjectType::NetworkPolicy => crate::objects::ObjectType::NetworkPolicy,
+            // This conversion is only exercised for `DefaultPrivilegesKey`. `ALTER DEFAULT
+            // PRIVILEGES .. ON METRIC SINKS` is rejected during planning before a
+            // `DefaultPrivilegeObject` naming a metric sink is constructed, so this arm is never
+            // reached.
+            ObjectType::MetricSink => {
+                unreachable!("metric sinks never appear in default privileges")
+            }
         }
     }
 
@@ -422,6 +437,13 @@ impl RustType<crate::objects::CommentObject> for CommentObjectId {
             CommentObjectId::Sink(global_id) => {
                 crate::objects::CommentObject::Sink(global_id.into_proto())
             }
+            // `CommentObjectId::MetricSink` is constructed on the drop path, but only to build a
+            // predicate that deletes any matching comment rows. It is never serialized into a
+            // durable `Comment` record, because `COMMENT ON METRIC SINK` isn't parseable
+            // (`CommentObjectType` has no `MetricSink` variant), so no such record ever exists.
+            CommentObjectId::MetricSink(_) => {
+                unreachable!("comments on metric sinks are never planned, so never serialized")
+            }
             CommentObjectId::Index(global_id) => {
                 crate::objects::CommentObject::Index(global_id.into_proto())
             }
diff --git a/src/catalog/src/durable/initialize.rs b/src/catalog/src/durable/initialize.rs
index f7011c0dc1a65..fb20c7e93168a 100644
--- a/src/catalog/src/durable/initialize.rs
+++ b/src/catalog/src/durable/initialize.rs
@@ -386,6 +386,9 @@ pub(crate) async fn initialize(
             ObjectType::MaterializedView => mz_audit_log::ObjectType::MaterializedView,
             ObjectType::Source => mz_audit_log::ObjectType::Source,
             ObjectType::Sink => mz_audit_log::ObjectType::Sink,
+            // Metric sinks are audited as sinks (see the `From<CatalogItemType>` impl in
+            // mz_sql::catalog) to avoid a durable audit-log proto bump.
+            ObjectType::MetricSink => mz_audit_log::ObjectType::Sink,
             ObjectType::Index => mz_audit_log::ObjectType::Index,
             ObjectType::Type => mz_audit_log::ObjectType::Type,
             ObjectType::Role => mz_audit_log::ObjectType::Role,
diff --git a/src/catalog/src/durable/objects.rs b/src/catalog/src/durable/objects.rs
index 6f57ae51eb3b4..d63d81f144f9e 100644
--- a/src/catalog/src/durable/objects.rs
+++ b/src/catalog/src/durable/objects.rs
@@ -1529,6 +1529,12 @@ pub fn item_type(create_sql: &str) -> CatalogItemType {
             assert_eq!(tokens.next(), Some("VIEW"));
             CatalogItemType::MaterializedView
         }
+        // Metric sinks are durably persisted like any other catalog item. This is the live
+        // parse path for their `create_sql`.
+        Some("METRIC") => {
+            assert_eq!(tokens.next(), Some("SINK"));
+            CatalogItemType::MetricSink
+        }
         Some("INDEX") => CatalogItemType::Index,
         Some("TYPE") => CatalogItemType::Type,
         Some("FUNCTION") => CatalogItemType::Func,
diff --git a/src/catalog/src/memory/objects.rs b/src/catalog/src/memory/objects.rs
index beae6a2c0aeb9..dfa5ad1b4bbcf 100644
--- a/src/catalog/src/memory/objects.rs
+++ b/src/catalog/src/memory/objects.rs
@@ -855,6 +855,7 @@ pub enum CatalogItem {
     Func(Func),
     Secret(Secret),
     Connection(Connection),
+    MetricSink(MetricSink),
 }
 
 impl From<CatalogEntry> for durable::Item {
@@ -1607,6 +1608,37 @@ impl Index {
     }
 }
 
+/// A metric sink, a compute object that exports a relation's rows as Prometheus metrics.
+#[derive(Debug, Clone, Serialize)]
+pub struct MetricSink {
+    /// Parse-able SQL that defines this metric sink.
+    pub create_sql: String,
+    /// [`GlobalId`] used to reference this metric sink from outside the catalog, e.g. compute.
+    pub global_id: GlobalId,
+    /// Collection we read into this metric sink.
+    pub from: GlobalId,
+    /// Other catalog objects referenced by this metric sink.
+    pub resolved_ids: ResolvedIds,
+    /// Cluster this metric sink runs on.
+    pub cluster_id: ClusterId,
+    /// Optimized global MIR plan, set after global optimization.
+    #[serde(skip)]
+    pub optimized_plan: Option<Arc<DataflowDescription<OptimizedMirRelationExpr>>>,
+    /// Physical (LIR) plan, set after physical optimization.
+    #[serde(skip)]
+    pub physical_plan: Option<Arc<DataflowDescription<ComputePlan>>>,
+    /// Dataflow metainfo (optimizer notices, etc.), set after optimization.
+    #[serde(skip)]
+    pub dataflow_metainfo: Option<DataflowMetainfo<Arc<OptimizerNotice>>>,
+}
+
+impl MetricSink {
+    /// The [`GlobalId`] that refers to this metric sink.
+    pub fn global_id(&self) -> GlobalId {
+        self.global_id
+    }
+}
+
 #[derive(Debug, Clone, Serialize)]
 pub struct Type {
     /// Parse-able SQL that defines this type.
@@ -1736,6 +1768,7 @@ impl CatalogItem {
             CatalogItem::Func(_) => CatalogItemType::Func,
             CatalogItem::Secret(_) => CatalogItemType::Secret,
             CatalogItem::Connection(_) => CatalogItemType::Connection,
+            CatalogItem::MetricSink(_) => CatalogItemType::MetricSink,
         }
     }
 
@@ -1754,6 +1787,7 @@ impl CatalogItem {
             CatalogItem::Type(ty) => ty.global_id,
             CatalogItem::Secret(secret) => secret.global_id,
             CatalogItem::Connection(conn) => conn.global_id,
+            CatalogItem::MetricSink(metric_sink) => metric_sink.global_id,
             CatalogItem::Table(table) => {
                 return itertools::Either::Left(table.collections.values().copied());
             }
@@ -1776,6 +1810,7 @@ impl CatalogItem {
             CatalogItem::Type(ty) => ty.global_id,
             CatalogItem::Secret(secret) => secret.global_id,
             CatalogItem::Connection(conn) => conn.global_id,
+            CatalogItem::MetricSink(metric_sink) => metric_sink.global_id,
             CatalogItem::Table(table) => table.global_id_writes(),
         }
     }
@@ -1785,6 +1820,7 @@ impl CatalogItem {
         match self {
             CatalogItem::Index(idx) => idx.optimized_plan.as_ref(),
             CatalogItem::MaterializedView(mv) => mv.optimized_plan.as_ref(),
+            CatalogItem::MetricSink(ms) => ms.optimized_plan.as_ref(),
             _ => None,
         }
     }
@@ -1794,6 +1830,7 @@ impl CatalogItem {
         match self {
             CatalogItem::Index(idx) => idx.physical_plan.as_ref(),
             CatalogItem::MaterializedView(mv) => mv.physical_plan.as_ref(),
+            CatalogItem::MetricSink(ms) => ms.physical_plan.as_ref(),
             _ => None,
         }
     }
@@ -1803,6 +1840,7 @@ impl CatalogItem {
         match self {
             CatalogItem::Index(idx) => idx.dataflow_metainfo.as_ref(),
             CatalogItem::MaterializedView(mv) => mv.dataflow_metainfo.as_ref(),
+            CatalogItem::MetricSink(ms) => ms.dataflow_metainfo.as_ref(),
             _ => None,
         }
     }
@@ -1829,6 +1867,11 @@ impl CatalogItem {
                 &mut mv.physical_plan,
                 &mut mv.dataflow_metainfo,
             )),
+            CatalogItem::MetricSink(ms) => Some((
+                &mut ms.optimized_plan,
+                &mut ms.physical_plan,
+                &mut ms.dataflow_metainfo,
+            )),
             _ => None,
         }
     }
@@ -1846,7 +1889,8 @@ impl CatalogItem {
             | CatalogItem::Type(_)
             | CatalogItem::Func(_)
             | CatalogItem::Secret(_)
-            | CatalogItem::Connection(_) => false,
+            | CatalogItem::Connection(_)
+            | CatalogItem::MetricSink(_) => false,
         }
     }
 
@@ -1872,7 +1916,8 @@ impl CatalogItem {
             | CatalogItem::Sink(_)
             | CatalogItem::Secret(_)
             | CatalogItem::Connection(_)
-            | CatalogItem::Type(_) => None,
+            | CatalogItem::Type(_)
+            | CatalogItem::MetricSink(_) => None,
         }
     }
 
@@ -1939,6 +1984,7 @@ impl CatalogItem {
             CatalogItem::MaterializedView(mview) => &mview.resolved_ids,
             CatalogItem::Secret(_) => &*EMPTY,
             CatalogItem::Connection(connection) => &connection.resolved_ids,
+            CatalogItem::MetricSink(metric_sink) => &metric_sink.resolved_ids,
         }
     }
 
@@ -1965,6 +2011,7 @@ impl CatalogItem {
             }
             CatalogItem::Secret(_) => {}
             CatalogItem::Connection(_) => {}
+            CatalogItem::MetricSink(_) => {}
         }
         uses
     }
@@ -1983,7 +2030,8 @@ impl CatalogItem {
             | CatalogItem::Secret(_)
             | CatalogItem::Type(_)
             | CatalogItem::Func(_)
-            | CatalogItem::Connection(_) => None,
+            | CatalogItem::Connection(_)
+            | CatalogItem::MetricSink(_) => None,
         }
     }
 
@@ -2001,7 +2049,8 @@ impl CatalogItem {
             | CatalogItem::Secret(_)
             | CatalogItem::Type(_)
             | CatalogItem::Func(_)
-            | CatalogItem::Connection(_) => (),
+            | CatalogItem::Connection(_)
+            | CatalogItem::MetricSink(_) => (),
         }
     }
 
@@ -2025,7 +2074,8 @@ impl CatalogItem {
             | CatalogItem::Secret(_)
             | CatalogItem::Type(_)
             | CatalogItem::Func(_)
-            | CatalogItem::Connection(_) => {
+            | CatalogItem::Connection(_)
+            | CatalogItem::MetricSink(_) => {
                 unreachable!("only views, indexes, and tables can be temporary")
             }
         }
@@ -2107,6 +2157,11 @@ impl CatalogItem {
                 Ok(CatalogItem::Type(i))
             }
             CatalogItem::Func(i) => Ok(CatalogItem::Func(i.clone())),
+            CatalogItem::MetricSink(i) => {
+                let mut i = i.clone();
+                i.create_sql = do_rewrite(i.create_sql)?;
+                Ok(CatalogItem::MetricSink(i))
+            }
         }
     }
 
@@ -2177,6 +2232,11 @@ impl CatalogItem {
                 i.create_sql = do_rewrite(i.create_sql)?;
                 Ok(CatalogItem::Connection(i))
             }
+            CatalogItem::MetricSink(i) => {
+                let mut i = i.clone();
+                i.create_sql = do_rewrite(i.create_sql)?;
+                Ok(CatalogItem::MetricSink(i))
+            }
         }
     }
 
@@ -2239,6 +2299,11 @@ impl CatalogItem {
                 i.create_sql = do_rewrite(i.create_sql);
                 CatalogItem::Connection(i)
             }
+            CatalogItem::MetricSink(i) => {
+                let mut i = i.clone();
+                i.create_sql = do_rewrite(i.create_sql);
+                CatalogItem::MetricSink(i)
+            }
         }
     }
     /// Updates the retain history for an item. Returns the previous retain history value. Returns
@@ -2416,7 +2481,8 @@ impl CatalogItem {
             | CatalogItem::MaterializedView(MaterializedView { create_sql, .. })
             | CatalogItem::Index(Index { create_sql, .. })
             | CatalogItem::Secret(Secret { create_sql, .. })
-            | CatalogItem::Connection(Connection { create_sql, .. }) => Some(create_sql),
+            | CatalogItem::Connection(Connection { create_sql, .. })
+            | CatalogItem::MetricSink(MetricSink { create_sql, .. }) => Some(create_sql),
             CatalogItem::Func(_) | CatalogItem::Log(_) => None,
         };
         let Some(create_sql) = create_sql else {
@@ -2442,6 +2508,7 @@ impl CatalogItem {
     pub fn is_compute_object_on_cluster(&self) -> Option<ClusterId> {
         match self {
             CatalogItem::Index(index) => Some(index.cluster_id),
+            CatalogItem::MetricSink(metric_sink) => Some(metric_sink.cluster_id),
             CatalogItem::Table(_)
             | CatalogItem::Source(_)
             | CatalogItem::Log(_)
@@ -2461,7 +2528,12 @@ impl CatalogItem {
     /// to a cluster but run no dataflow on any replica.
     pub fn is_hydratable(&self) -> bool {
         match self {
-            CatalogItem::Index(_) | CatalogItem::MaterializedView(_) | CatalogItem::Sink(_) => true,
+            CatalogItem::Index(_)
+            | CatalogItem::MaterializedView(_)
+            | CatalogItem::Sink(_)
+            // A metric sink runs a dataflow on its cluster's replicas, so it has hydration state
+            // like an index or sink.
+            | CatalogItem::MetricSink(_) => true,
             CatalogItem::Source(source) => matches!(
                 source.data_source,
                 DataSourceDesc::Ingestion { .. } | DataSourceDesc::OldSyntaxIngestion { .. }
@@ -2480,6 +2552,7 @@ impl CatalogItem {
         match self {
             CatalogItem::MaterializedView(mv) => Some(mv.cluster_id),
             CatalogItem::Index(index) => Some(index.cluster_id),
+            CatalogItem::MetricSink(metric_sink) => Some(metric_sink.cluster_id),
             CatalogItem::Source(source) => match &source.data_source {
                 DataSourceDesc::Ingestion { cluster_id, .. }
                 | DataSourceDesc::OldSyntaxIngestion { cluster_id, .. } => Some(*cluster_id),
@@ -2517,7 +2590,8 @@ impl CatalogItem {
             | CatalogItem::Type(_)
             | CatalogItem::Func(_)
             | CatalogItem::Secret(_)
-            | CatalogItem::Connection(_) => None,
+            | CatalogItem::Connection(_)
+            | CatalogItem::MetricSink(_) => None,
         }
     }
 
@@ -2538,7 +2612,8 @@ impl CatalogItem {
             | CatalogItem::Type(_)
             | CatalogItem::Func(_)
             | CatalogItem::Secret(_)
-            | CatalogItem::Connection(_) => return None,
+            | CatalogItem::Connection(_)
+            | CatalogItem::MetricSink(_) => return None,
         };
         Some(cw)
     }
@@ -2562,7 +2637,8 @@ impl CatalogItem {
             | CatalogItem::Type(_)
             | CatalogItem::Func(_)
             | CatalogItem::Secret(_)
-            | CatalogItem::Connection(_) => return None,
+            | CatalogItem::Connection(_)
+            | CatalogItem::MetricSink(_) => return None,
         };
         Some(custom_logical_compaction_window.unwrap_or(CompactionWindow::Default))
     }
@@ -2582,7 +2658,8 @@ impl CatalogItem {
             | CatalogItem::Type(_)
             | CatalogItem::Func(_)
             | CatalogItem::Secret(_)
-            | CatalogItem::Connection(_) => false,
+            | CatalogItem::Connection(_)
+            | CatalogItem::MetricSink(_) => false,
         }
     }
 
@@ -2639,6 +2716,7 @@ impl CatalogItem {
                 BTreeMap::new(),
             ),
             CatalogItem::Func(_) => unreachable!("cannot serialize functions yet"),
+            CatalogItem::MetricSink(ms) => (ms.create_sql.clone(), ms.global_id, BTreeMap::new()),
         }
     }
 
@@ -2684,6 +2762,7 @@ impl CatalogItem {
                 (connection.create_sql, connection.global_id, BTreeMap::new())
             }
             CatalogItem::Func(_) => unreachable!("cannot serialize functions yet"),
+            CatalogItem::MetricSink(ms) => (ms.create_sql, ms.global_id, BTreeMap::new()),
         }
     }
 
@@ -2702,6 +2781,7 @@ impl CatalogItem {
             CatalogItem::Func(func) => return Some(func.global_id),
             CatalogItem::Secret(secret) => return Some(secret.global_id),
             CatalogItem::Connection(conn) => return Some(conn.global_id),
+            CatalogItem::MetricSink(metric_sink) => return Some(metric_sink.global_id),
         };
         match version {
             RelationVersionSelector::Latest => collections.values().last().copied(),
@@ -2906,7 +2986,8 @@ impl CatalogEntry {
             | CatalogItem::Type(_)
             | CatalogItem::Func(_)
             | CatalogItem::Secret(_)
-            | CatalogItem::Connection(_) => None,
+            | CatalogItem::Connection(_)
+            | CatalogItem::MetricSink(_) => None,
         }
     }
 
@@ -2940,6 +3021,11 @@ impl CatalogEntry {
         matches!(self.item(), CatalogItem::Index(_))
     }
 
+    /// Reports whether this catalog entry is a metric sink.
+    pub fn is_metric_sink(&self) -> bool {
+        matches!(self.item(), CatalogItem::MetricSink(_))
+    }
+
     /// Reports whether this catalog entry can be treated as a relation, it can produce rows.
     pub fn is_relation(&self) -> bool {
         mz_sql::catalog::ObjectType::from(self.item_type()).is_relation()
@@ -3035,6 +3121,7 @@ impl CatalogEntry {
             Connection => CommentObjectId::Connection(self.id),
             Type => CommentObjectId::Type(self.id),
             Secret => CommentObjectId::Secret(self.id),
+            MetricSink => CommentObjectId::MetricSink(self.id),
         }
     }
 }
@@ -4023,6 +4110,7 @@ impl mz_sql::catalog::CatalogItem for CatalogEntry {
             }
             CatalogItem::Secret(Secret { create_sql, .. }) => create_sql,
             CatalogItem::Connection(Connection { create_sql, .. }) => create_sql,
+            CatalogItem::MetricSink(MetricSink { create_sql, .. }) => create_sql,
             CatalogItem::Func(_) => "<builtin>",
             CatalogItem::Log(_) => "<builtin>",
         }
diff --git a/src/compute-client/src/controller.rs b/src/compute-client/src/controller.rs
index 33c3d9a1a2672..a76d40614cf72 100644
--- a/src/compute-client/src/controller.rs
+++ b/src/compute-client/src/controller.rs
@@ -778,7 +778,9 @@ impl ComputeController {
 
     /// Creates the described dataflow and initializes state for its output.
     ///
-    /// Only materialized views and subscribes are allowed to have a `target_replica`.
+    /// Only materialized views, subscribes, and metric sinks are allowed to have a
+    /// `target_replica`. A metric sink is a sink export, not an index export, so it passes the
+    /// index-export assertion below.
     ///
     /// Panics if called with a dataflow description that has index exports
     /// when `target_replica` is set.
diff --git a/src/environmentd/src/http/sql.rs b/src/environmentd/src/http/sql.rs
index 9e07c81f27f90..360eeb5da1e62 100644
--- a/src/environmentd/src/http/sql.rs
+++ b/src/environmentd/src/http/sql.rs
@@ -1527,6 +1527,7 @@ async fn execute_stmt<S: ResultSender>(
         | ExecuteResponse::CreatedClusterReplica { .. }
         | ExecuteResponse::CreatedTable { .. }
         | ExecuteResponse::CreatedIndex { .. }
+        | ExecuteResponse::CreatedMetricSink { .. }
         | ExecuteResponse::CreatedIntrospectionSubscribe
         | ExecuteResponse::CreatedSecret { .. }
         | ExecuteResponse::CreatedSource { .. }
diff --git a/src/pgwire/src/protocol.rs b/src/pgwire/src/protocol.rs
index 58cd2fa7785e9..b04030204d93a 100644
--- a/src/pgwire/src/protocol.rs
+++ b/src/pgwire/src/protocol.rs
@@ -2394,6 +2394,7 @@ where
             | ExecuteResponse::CreatedConnection { .. }
             | ExecuteResponse::CreatedDatabase { .. }
             | ExecuteResponse::CreatedIndex { .. }
+            | ExecuteResponse::CreatedMetricSink { .. }
             | ExecuteResponse::CreatedIntrospectionSubscribe
             | ExecuteResponse::CreatedMaterializedView { .. }
             | ExecuteResponse::CreatedRole
diff --git a/src/sql-lexer/src/keywords.txt b/src/sql-lexer/src/keywords.txt
index 12c574af645d2..20eff3d836bae 100644
--- a/src/sql-lexer/src/keywords.txt
+++ b/src/sql-lexer/src/keywords.txt
@@ -297,6 +297,7 @@ Membership
 Memory
 Message
 Metadata
+Metric
 Minute
 Minutes
 Mock
diff --git a/src/sql-parser/src/ast/defs/statement.rs b/src/sql-parser/src/ast/defs/statement.rs
index a131594360f3e..13e759b1edea5 100644
--- a/src/sql-parser/src/ast/defs/statement.rs
+++ b/src/sql-parser/src/ast/defs/statement.rs
@@ -53,6 +53,7 @@ pub enum Statement<T: AstInfo> {
     CreateSource(CreateSourceStatement<T>),
     CreateSubsource(CreateSubsourceStatement<T>),
     CreateSink(CreateSinkStatement<T>),
+    CreateMetricSink(CreateMetricSinkStatement<T>),
     CreateView(CreateViewStatement<T>),
     CreateMaterializedView(CreateMaterializedViewStatement<T>),
     CreateTable(CreateTableStatement<T>),
@@ -132,6 +133,7 @@ impl<T: AstInfo> AstDisplay for Statement<T> {
             Statement::CreateSource(stmt) => f.write_node(stmt),
             Statement::CreateSubsource(stmt) => f.write_node(stmt),
             Statement::CreateSink(stmt) => f.write_node(stmt),
+            Statement::CreateMetricSink(stmt) => f.write_node(stmt),
             Statement::CreateView(stmt) => f.write_node(stmt),
             Statement::CreateMaterializedView(stmt) => f.write_node(stmt),
             Statement::CreateTable(stmt) => f.write_node(stmt),
@@ -239,6 +241,7 @@ pub fn statement_kind_label_value(kind: StatementKind) -> &'static str {
         StatementKind::CreateSource => "create_source",
         StatementKind::CreateSubsource => "create_subsource",
         StatementKind::CreateSink => "create_sink",
+        StatementKind::CreateMetricSink => "create_metric_sink",
         StatementKind::CreateView => "create_view",
         StatementKind::CreateMaterializedView => "create_materialized_view",
         StatementKind::CreateTable => "create_table",
@@ -1432,6 +1435,36 @@ impl<T: AstInfo> AstDisplay for CreateSinkStatement<T> {
 }
 impl_display_t!(CreateSinkStatement);
 
+/// `CREATE METRIC SINK`
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct CreateMetricSinkStatement<T: AstInfo> {
+    pub name: Option<UnresolvedItemName>,
+    pub in_cluster: Option<T::ClusterName>,
+    pub if_not_exists: bool,
+    pub from: T::ItemName,
+}
+
+impl<T: AstInfo> AstDisplay for CreateMetricSinkStatement<T> {
+    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
+        f.write_str("CREATE METRIC SINK ");
+        if self.if_not_exists {
+            f.write_str("IF NOT EXISTS ");
+        }
+        if let Some(name) = &self.name {
+            f.write_node(name);
+            f.write_str(" ");
+        }
+        if let Some(cluster) = &self.in_cluster {
+            f.write_str("IN CLUSTER ");
+            f.write_node(cluster);
+            f.write_str(" ");
+        }
+        f.write_str("FROM ");
+        f.write_node(&self.from);
+    }
+}
+impl_display_t!(CreateMetricSinkStatement);
+
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct ViewDefinition<T: AstInfo> {
     /// View name
@@ -4341,6 +4374,7 @@ pub enum ObjectType {
     MaterializedView,
     Source,
     Sink,
+    MetricSink,
     Index,
     Type,
     Role,
@@ -4363,6 +4397,7 @@ impl ObjectType {
             | ObjectType::MaterializedView
             | ObjectType::Source
             | ObjectType::Sink
+            | ObjectType::MetricSink
             | ObjectType::Index
             | ObjectType::Type
             | ObjectType::Secret
@@ -4387,6 +4422,7 @@ impl AstDisplay for ObjectType {
             ObjectType::MaterializedView => "MATERIALIZED VIEW",
             ObjectType::Source => "SOURCE",
             ObjectType::Sink => "SINK",
+            ObjectType::MetricSink => "METRIC SINK",
             ObjectType::Index => "INDEX",
             ObjectType::Type => "TYPE",
             ObjectType::Role => "ROLE",
diff --git a/src/sql-parser/src/parser.rs b/src/sql-parser/src/parser.rs
index e0c2a39f779eb..cb685a6e6ea5c 100644
--- a/src/sql-parser/src/parser.rs
+++ b/src/sql-parser/src/parser.rs
@@ -2060,6 +2060,9 @@ impl<'a> Parser<'a> {
         } else if self.peek_keyword(SCHEMA) {
             self.parse_create_schema()
                 .map_parser_err(StatementKind::CreateSchema)
+        } else if self.peek_keywords(&[METRIC, SINK]) {
+            self.parse_create_metric_sink()
+                .map_parser_err(StatementKind::CreateMetricSink)
         } else if self.peek_keyword(SINK) {
             self.parse_create_sink()
                 .map_parser_err(StatementKind::CreateSink)
@@ -3641,6 +3644,21 @@ impl<'a> Parser<'a> {
         Ok(Statement::CreateSink(statement))
     }
 
+    fn parse_create_metric_sink(&mut self) -> Result<Statement<Raw>, ParserError> {
+        self.expect_keywords(&[METRIC, SINK])?;
+        let if_not_exists = self.parse_if_not_exists()?;
+        let name = Some(self.parse_item_name()?);
+        let in_cluster = self.parse_optional_in_cluster()?;
+        self.expect_keyword(FROM)?;
+        let from = self.parse_raw_name()?;
+        Ok(Statement::CreateMetricSink(CreateMetricSinkStatement {
+            name,
+            in_cluster,
+            if_not_exists,
+            from,
+        }))
+    }
+
     /// Parse the name of a CREATE SINK optional parameter
     fn parse_create_sink_option_name(&mut self) -> Result<CreateSinkOptionName, ParserError> {
         let name = match self.expect_one_of_keywords(&[PARTITION, SNAPSHOT, VERSION, COMMIT])? {
@@ -5039,6 +5057,7 @@ impl<'a> Parser<'a> {
             | ObjectType::MaterializedView
             | ObjectType::Source
             | ObjectType::Sink
+            | ObjectType::MetricSink
             | ObjectType::Index
             | ObjectType::Type
             | ObjectType::Secret
@@ -5880,7 +5899,7 @@ impl<'a> Parser<'a> {
             ObjectType::NetworkPolicy => self
                 .parse_alter_network_policy()
                 .map_parser_err(StatementKind::AlterNetworkPolicy),
-            ObjectType::Func | ObjectType::Subsource => parser_err!(
+            ObjectType::Func | ObjectType::Subsource | ObjectType::MetricSink => parser_err!(
                 self,
                 self.peek_prev_pos(),
                 format!("Unsupported ALTER on {object_type}")
@@ -6629,6 +6648,7 @@ impl<'a> Parser<'a> {
             ObjectType::View => &[SET, RENAME, OWNER, RESET],
             ObjectType::Source
             | ObjectType::Sink
+            | ObjectType::MetricSink
             | ObjectType::Index
             | ObjectType::Type
             | ObjectType::Role
@@ -7475,6 +7495,7 @@ impl<'a> Parser<'a> {
             | ObjectType::Source
             | ObjectType::Subsource
             | ObjectType::Sink
+            | ObjectType::MetricSink
             | ObjectType::Index
             | ObjectType::Type
             | ObjectType::Secret
@@ -8299,7 +8320,7 @@ impl<'a> Parser<'a> {
                         on_object,
                     }
                 }
-                ObjectType::Func => {
+                ObjectType::Func | ObjectType::MetricSink => {
                     return parser_err!(
                         self,
                         self.peek_prev_pos(),
@@ -9876,6 +9897,7 @@ impl<'a> Parser<'a> {
                 )
             }
             ObjectType::Sink
+            | ObjectType::MetricSink
             | ObjectType::Index
             | ObjectType::ClusterReplica
             | ObjectType::Role
@@ -9907,6 +9929,7 @@ impl<'a> Parser<'a> {
                 MATERIALIZED,
                 SOURCE,
                 SINK,
+                METRIC,
                 INDEX,
                 TYPE,
                 ROLE,
@@ -9930,6 +9953,13 @@ impl<'a> Parser<'a> {
                 }
                 SOURCE => ObjectType::Source,
                 SINK => ObjectType::Sink,
+                METRIC => {
+                    if let Err(e) = self.expect_keyword(SINK) {
+                        self.prev_token();
+                        return Err(e);
+                    }
+                    ObjectType::MetricSink
+                }
                 INDEX => ObjectType::Index,
                 TYPE => ObjectType::Type,
                 ROLE | USER => ObjectType::Role,
@@ -9966,6 +9996,7 @@ impl<'a> Parser<'a> {
                 MATERIALIZED,
                 SOURCE,
                 SINK,
+                METRIC,
                 INDEX,
                 TYPE,
                 ROLE,
@@ -9990,6 +10021,14 @@ impl<'a> Parser<'a> {
                 }
                 SOURCE => ObjectType::Source,
                 SINK => ObjectType::Sink,
+                METRIC => {
+                    if self.parse_keyword(SINK) {
+                        ObjectType::MetricSink
+                    } else {
+                        self.prev_token();
+                        return None;
+                    }
+                }
                 INDEX => ObjectType::Index,
                 TYPE => ObjectType::Type,
                 ROLE | USER => ObjectType::Role,
diff --git a/src/sql-parser/tests/testdata/ddl b/src/sql-parser/tests/testdata/ddl
index 62597942f6bf5..60771e7c1a5b9 100644
--- a/src/sql-parser/tests/testdata/ddl
+++ b/src/sql-parser/tests/testdata/ddl
@@ -1103,6 +1103,20 @@ CREATE SINK bar FROM foo INTO ICEBERG CATALOG CONNECTION s3tables (NAMESPACE = '
 =>
 CreateSink(CreateSinkStatement { name: Some(UnresolvedItemName([Ident("bar")])), in_cluster: None, if_not_exists: false, from: Name(UnresolvedItemName([Ident("foo")])), connection: Iceberg { catalog_connection: Name(UnresolvedItemName([Ident("s3tables")])), aws_connection: Some(Name(UnresolvedItemName([Ident("aws_conn")]))), key: None, options: [IcebergSinkConfigOption { name: Namespace, value: Some(Value(String("testnamespace"))) }, IcebergSinkConfigOption { name: Table, value: Some(Value(String("daily_sales"))) }] }, format: None, envelope: None, mode: Some(Append), with_options: [] })
 
+parse-statement
+CREATE METRIC SINK m IN CLUSTER c FROM v
+----
+CREATE METRIC SINK m IN CLUSTER c FROM v
+=>
+CreateMetricSink(CreateMetricSinkStatement { name: Some(UnresolvedItemName([Ident("m")])), in_cluster: Some(Unresolved(Ident("c"))), if_not_exists: false, from: Name(UnresolvedItemName([Ident("v")])) })
+
+parse-statement
+CREATE METRIC SINK IF NOT EXISTS m FROM v
+----
+CREATE METRIC SINK IF NOT EXISTS m FROM v
+=>
+CreateMetricSink(CreateMetricSinkStatement { name: Some(UnresolvedItemName([Ident("m")])), in_cluster: None, if_not_exists: true, from: Name(UnresolvedItemName([Ident("v")])) })
+
 parse-statement
 CREATE INDEX foo ON myschema.bar (a, b)
 ----
@@ -1327,6 +1341,13 @@ DROP SOURCE myschema.mydatasource
 =>
 DropObjects(DropObjectsStatement { object_type: Source, if_exists: false, names: [Item(UnresolvedItemName([Ident("myschema"), Ident("mydatasource")]))], cascade: false })
 
+parse-statement
+DROP METRIC SINK m
+----
+DROP METRIC SINK m
+=>
+DropObjects(DropObjectsStatement { object_type: MetricSink, if_exists: false, names: [Item(UnresolvedItemName([Ident("m")]))], cascade: false })
+
 parse-statement
 DROP INDEX IF EXISTS myschema.myindex
 ----
diff --git a/src/sql/src/catalog.rs b/src/sql/src/catalog.rs
index 9806b73ccc2fc..0039e1606403a 100644
--- a/src/sql/src/catalog.rs
+++ b/src/sql/src/catalog.rs
@@ -962,6 +962,8 @@ pub enum CatalogItemType {
     Secret,
     /// A connection.
     Connection,
+    /// A metric sink.
+    MetricSink,
 }
 
 impl CatalogItemType {
@@ -995,6 +997,7 @@ impl CatalogItemType {
             CatalogItemType::Func => false,
             CatalogItemType::Secret => false,
             CatalogItemType::Connection => false,
+            CatalogItemType::MetricSink => false,
         }
     }
 }
@@ -1012,6 +1015,7 @@ impl fmt::Display for CatalogItemType {
             CatalogItemType::Func => f.write_str("func"),
             CatalogItemType::Secret => f.write_str("secret"),
             CatalogItemType::Connection => f.write_str("connection"),
+            CatalogItemType::MetricSink => f.write_str("metric sink"),
         }
     }
 }
@@ -1029,6 +1033,7 @@ impl From<CatalogItemType> for ObjectType {
             CatalogItemType::Func => ObjectType::Func,
             CatalogItemType::Secret => ObjectType::Secret,
             CatalogItemType::Connection => ObjectType::Connection,
+            CatalogItemType::MetricSink => ObjectType::MetricSink,
         }
     }
 }
@@ -1046,6 +1051,10 @@ impl From<CatalogItemType> for mz_audit_log::ObjectType {
             CatalogItemType::Func => mz_audit_log::ObjectType::Func,
             CatalogItemType::Secret => mz_audit_log::ObjectType::Secret,
             CatalogItemType::Connection => mz_audit_log::ObjectType::Connection,
+            // Metric sinks are audited as sinks. Keeping them out of the audit-log
+            // `ObjectType` enum avoids a durable audit-log proto bump (they persist as ordinary
+            // items keyed on `create_sql`; only the append-only audit log would need the variant).
+            CatalogItemType::MetricSink => mz_audit_log::ObjectType::Sink,
         }
     }
 }
@@ -1563,6 +1572,7 @@ pub enum ObjectType {
     MaterializedView,
     Source,
     Sink,
+    MetricSink,
     Index,
     Type,
     Role,
@@ -1585,6 +1595,7 @@ impl ObjectType {
             | ObjectType::MaterializedView
             | ObjectType::Source => true,
             ObjectType::Sink
+            | ObjectType::MetricSink
             | ObjectType::Index
             | ObjectType::Type
             | ObjectType::Secret
@@ -1609,6 +1620,7 @@ impl From<mz_sql_parser::ast::ObjectType> for ObjectType {
             mz_sql_parser::ast::ObjectType::Source => ObjectType::Source,
             mz_sql_parser::ast::ObjectType::Subsource => ObjectType::Source,
             mz_sql_parser::ast::ObjectType::Sink => ObjectType::Sink,
+            mz_sql_parser::ast::ObjectType::MetricSink => ObjectType::MetricSink,
             mz_sql_parser::ast::ObjectType::Index => ObjectType::Index,
             mz_sql_parser::ast::ObjectType::Type => ObjectType::Type,
             mz_sql_parser::ast::ObjectType::Role => ObjectType::Role,
@@ -1632,6 +1644,7 @@ impl From<CommentObjectId> for ObjectType {
             CommentObjectId::MaterializedView(_) => ObjectType::MaterializedView,
             CommentObjectId::Source(_) => ObjectType::Source,
             CommentObjectId::Sink(_) => ObjectType::Sink,
+            CommentObjectId::MetricSink(_) => ObjectType::MetricSink,
             CommentObjectId::Index(_) => ObjectType::Index,
             CommentObjectId::Func(_) => ObjectType::Func,
             CommentObjectId::Connection(_) => ObjectType::Connection,
@@ -1655,6 +1668,7 @@ impl Display for ObjectType {
             ObjectType::MaterializedView => "MATERIALIZED VIEW",
             ObjectType::Source => "SOURCE",
             ObjectType::Sink => "SINK",
+            ObjectType::MetricSink => "METRIC SINK",
             ObjectType::Index => "INDEX",
             ObjectType::Type => "TYPE",
             ObjectType::Role => "ROLE",
diff --git a/src/sql/src/names.rs b/src/sql/src/names.rs
index 3d9bddbeb3aaa..ca07fa73c47a5 100644
--- a/src/sql/src/names.rs
+++ b/src/sql/src/names.rs
@@ -1232,6 +1232,7 @@ impl From<CommentObjectId> for ObjectId {
             | 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)
@@ -1292,6 +1293,7 @@ pub enum CommentObjectId {
     MaterializedView(CatalogItemId),
     Source(CatalogItemId),
     Sink(CatalogItemId),
+    MetricSink(CatalogItemId),
     Index(CatalogItemId),
     Func(CatalogItemId),
     Connection(CatalogItemId),
diff --git a/src/sql/src/normalize.rs b/src/sql/src/normalize.rs
index f4da9520ed414..5c0c2afb37f35 100644
--- a/src/sql/src/normalize.rs
+++ b/src/sql/src/normalize.rs
@@ -22,11 +22,11 @@ use mz_sql_parser::ast::display::AstDisplay;
 use mz_sql_parser::ast::visit_mut::{self, VisitMut};
 use mz_sql_parser::ast::{
     CreateConnectionStatement, CreateIndexStatement, CreateMaterializedViewStatement,
-    CreateSecretStatement, CreateSinkStatement, CreateSourceStatement, CreateSubsourceStatement,
-    CreateTableFromSourceStatement, CreateTableStatement, CreateTypeStatement, CreateViewStatement,
-    CreateWebhookSourceStatement, CteBlock, Function, FunctionArgs, Ident, IfExistsBehavior,
-    MutRecBlock, Op, Query, Statement, TableFactor, TableFromSourceColumns, UnresolvedItemName,
-    UnresolvedSchemaName, Value, ViewDefinition,
+    CreateMetricSinkStatement, CreateSecretStatement, CreateSinkStatement, CreateSourceStatement,
+    CreateSubsourceStatement, CreateTableFromSourceStatement, CreateTableStatement,
+    CreateTypeStatement, CreateViewStatement, CreateWebhookSourceStatement, CteBlock, Function,
+    FunctionArgs, Ident, IfExistsBehavior, MutRecBlock, Op, Query, Statement, TableFactor,
+    TableFromSourceColumns, UnresolvedItemName, UnresolvedSchemaName, Value, ViewDefinition,
 };
 
 use crate::names::{Aug, FullItemName, PartialItemName, PartialSchemaName, RawDatabaseSpecifier};
@@ -375,6 +375,18 @@ pub fn create_statement(
             *if_not_exists = false;
         }
 
+        Statement::CreateMetricSink(CreateMetricSinkStatement {
+            name,
+            in_cluster: _,
+            if_not_exists,
+            from: _,
+        }) => {
+            if let Some(name) = name {
+                *name = allocate_name(name)?;
+            }
+            *if_not_exists = false;
+        }
+
         Statement::CreateView(CreateViewStatement {
             temporary,
             if_exists,
diff --git a/src/sql/src/plan.rs b/src/sql/src/plan.rs
index a5981627894e2..4153b51af1b20 100644
--- a/src/sql/src/plan.rs
+++ b/src/sql/src/plan.rs
@@ -147,6 +147,7 @@ pub enum Plan {
     CreateMaterializedView(CreateMaterializedViewPlan),
     CreateNetworkPolicy(CreateNetworkPolicyPlan),
     CreateIndex(CreateIndexPlan),
+    CreateMetricSink(CreateMetricSinkPlan),
     CreateType(CreateTypePlan),
     Comment(CommentPlan),
     DiscardTemp,
@@ -280,6 +281,7 @@ impl Plan {
             StatementKind::CreateSchema => &[PlanKind::CreateSchema],
             StatementKind::CreateSecret => &[PlanKind::CreateSecret],
             StatementKind::CreateSink => &[PlanKind::CreateSink],
+            StatementKind::CreateMetricSink => &[PlanKind::CreateMetricSink],
             StatementKind::CreateSource | StatementKind::CreateSubsource => {
                 &[PlanKind::CreateSource]
             }
@@ -349,6 +351,7 @@ impl Plan {
             Plan::CreateView(_) => "create view",
             Plan::CreateMaterializedView(_) => "create materialized view",
             Plan::CreateIndex(_) => "create index",
+            Plan::CreateMetricSink(_) => "create metric sink",
             Plan::CreateType(_) => "create type",
             Plan::CreateNetworkPolicy(_) => "create network policy",
             Plan::Comment(_) => "comment",
@@ -360,6 +363,7 @@ impl Plan {
                 ObjectType::MaterializedView => "drop materialized view",
                 ObjectType::Source => "drop source",
                 ObjectType::Sink => "drop sink",
+                ObjectType::MetricSink => "drop metric sink",
                 ObjectType::Index => "drop index",
                 ObjectType::Type => "drop type",
                 ObjectType::Role => "drop roles",
@@ -400,6 +404,7 @@ impl Plan {
                 ObjectType::MaterializedView => "alter materialized view",
                 ObjectType::Source => "alter source",
                 ObjectType::Sink => "alter sink",
+                ObjectType::MetricSink => "alter metric sink",
                 ObjectType::Index => "alter index",
                 ObjectType::Type => "alter type",
                 ObjectType::Role => "alter role",
@@ -435,6 +440,7 @@ impl Plan {
                 ObjectType::MaterializedView => "alter materialized view owner",
                 ObjectType::Source => "alter source owner",
                 ObjectType::Sink => "alter sink owner",
+                ObjectType::MetricSink => "alter metric sink owner",
                 ObjectType::Index => "alter index owner",
                 ObjectType::Type => "alter type owner",
                 ObjectType::Role => "alter role owner",
@@ -808,6 +814,14 @@ pub struct CreateIndexPlan {
     pub if_not_exists: bool,
 }
 
+#[derive(Debug, Clone)]
+pub struct CreateMetricSinkPlan {
+    pub name: QualifiedItemName,
+    pub metric_sink: MetricSink,
+    pub if_not_exists: bool,
+    pub in_cluster: ClusterId,
+}
+
 #[derive(Debug)]
 pub struct CreateTypePlan {
     pub name: QualifiedItemName,
@@ -1977,6 +1991,15 @@ pub struct Index {
     pub cluster_id: ClusterId,
 }
 
+#[derive(Clone, Debug)]
+pub struct MetricSink {
+    /// Parse-able SQL that defines this metric sink.
+    pub create_sql: String,
+    /// Collection we read into this metric sink.
+    pub from: GlobalId,
+    pub cluster_id: ClusterId,
+}
+
 #[derive(Clone, Debug)]
 pub struct Type {
     pub create_sql: String,
diff --git a/src/sql/src/plan/statement.rs b/src/sql/src/plan/statement.rs
index d69cdac72c989..50fa57a7d446d 100644
--- a/src/sql/src/plan/statement.rs
+++ b/src/sql/src/plan/statement.rs
@@ -152,6 +152,7 @@ pub fn describe(
         Statement::CreateSchema(stmt) => ddl::describe_create_schema(&scx, stmt)?,
         Statement::CreateSecret(stmt) => ddl::describe_create_secret(&scx, stmt)?,
         Statement::CreateSink(stmt) => ddl::describe_create_sink(&scx, stmt)?,
+        Statement::CreateMetricSink(stmt) => ddl::describe_create_metric_sink(&scx, stmt)?,
         Statement::CreateWebhookSource(stmt) => ddl::describe_create_webhook_source(&scx, stmt)?,
         Statement::CreateSource(stmt) => ddl::describe_create_source(&scx, stmt)?,
         Statement::CreateSubsource(stmt) => ddl::describe_create_subsource(&scx, stmt)?,
@@ -357,6 +358,7 @@ pub fn plan(
         Statement::CreateSchema(stmt) => ddl::plan_create_schema(scx, stmt),
         Statement::CreateSecret(stmt) => ddl::plan_create_secret(scx, stmt),
         Statement::CreateSink(stmt) => ddl::plan_create_sink(scx, stmt),
+        Statement::CreateMetricSink(stmt) => ddl::plan_create_metric_sink(scx, stmt),
         Statement::CreateWebhookSource(stmt) => ddl::plan_create_webhook_source(scx, stmt),
         Statement::CreateSource(stmt) => ddl::plan_create_source(scx, stmt),
         Statement::CreateSubsource(stmt) => ddl::plan_create_subsource(scx, stmt),
@@ -502,6 +504,7 @@ impl PartialEq<ObjectType> for CatalogItemType {
             (CatalogItemType::Source, ObjectType::Source)
             | (CatalogItemType::Table, ObjectType::Table)
             | (CatalogItemType::Sink, ObjectType::Sink)
+            | (CatalogItemType::MetricSink, ObjectType::MetricSink)
             | (CatalogItemType::View, ObjectType::View)
             | (CatalogItemType::MaterializedView, ObjectType::MaterializedView)
             | (CatalogItemType::Index, ObjectType::Index)
@@ -1080,6 +1083,7 @@ impl<T: mz_sql_parser::ast::AstInfo> From<&Statement<T>> for StatementClassifica
             Statement::CreateSchema(_) => DDL,
             Statement::CreateSecret(_) => DDL,
             Statement::CreateSink(_) => DDL,
+            Statement::CreateMetricSink(_) => DDL,
             Statement::CreateWebhookSource(_) => DDL,
             Statement::CreateSource(_) => DDL,
             Statement::CreateSubsource(_) => DDL,
diff --git a/src/sql/src/plan/statement/acl.rs b/src/sql/src/plan/statement/acl.rs
index 8ebeb82038702..eaa9f85a4c64a 100644
--- a/src/sql/src/plan/statement/acl.rs
+++ b/src/sql/src/plan/statement/acl.rs
@@ -669,7 +669,11 @@ pub fn plan_alter_default_privileges(
         ObjectType::View | ObjectType::MaterializedView | ObjectType::Source => sql_bail!(
             "{object_type}S is not valid for ALTER DEFAULT PRIVILEGES, use TABLES instead"
         ),
-        ObjectType::Sink | ObjectType::ClusterReplica | ObjectType::Role | ObjectType::Func => {
+        ObjectType::Sink
+        | ObjectType::MetricSink
+        | ObjectType::ClusterReplica
+        | ObjectType::Role
+        | ObjectType::Func => {
             sql_bail!("{object_type}S do not have privileges")
         }
         ObjectType::Cluster | ObjectType::Database
diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs
index 0f57c6f2ca5dd..03e5bb19536d9 100644
--- a/src/sql/src/plan/statement/ddl.rs
+++ b/src/sql/src/plan/statement/ddl.rs
@@ -58,10 +58,10 @@ use mz_sql_parser::ast::{
     CommentStatement, ConnectionOption, ConnectionOptionName, CreateClusterReplicaStatement,
     CreateClusterStatement, CreateConnectionOption, CreateConnectionOptionName,
     CreateConnectionStatement, CreateConnectionType, CreateDatabaseStatement, CreateIndexStatement,
-    CreateMaterializedViewStatement, CreateNetworkPolicyStatement, CreateRoleStatement,
-    CreateSchemaStatement, CreateSecretStatement, CreateSinkConnection, CreateSinkOption,
-    CreateSinkOptionName, CreateSinkStatement, CreateSourceConnection, CreateSourceOption,
-    CreateSourceOptionName, CreateSourceStatement, CreateSubsourceOption,
+    CreateMaterializedViewStatement, CreateMetricSinkStatement, CreateNetworkPolicyStatement,
+    CreateRoleStatement, CreateSchemaStatement, CreateSecretStatement, CreateSinkConnection,
+    CreateSinkOption, CreateSinkOptionName, CreateSinkStatement, CreateSourceConnection,
+    CreateSourceOption, CreateSourceOptionName, CreateSourceStatement, CreateSubsourceOption,
     CreateSubsourceOptionName, CreateSubsourceStatement, CreateTableFromSourceStatement,
     CreateTableStatement, CreateTypeAs, CreateTypeListOption, CreateTypeListOptionName,
     CreateTypeMapOption, CreateTypeMapOptionName, CreateTypeStatement, CreateViewStatement,
@@ -156,19 +156,20 @@ use crate::plan::{
     ComputeReplicaConfig, ComputeReplicaIntrospectionConfig, ConnectionDetails,
     CreateClusterManagedPlan, CreateClusterPlan, CreateClusterReplicaPlan,
     CreateClusterUnmanagedPlan, CreateClusterVariant, CreateConnectionPlan, CreateDatabasePlan,
-    CreateIndexPlan, CreateMaterializedViewPlan, CreateNetworkPolicyPlan, CreateRolePlan,
-    CreateSchemaPlan, CreateSecretPlan, CreateSinkPlan, CreateSourcePlan, CreateTablePlan,
-    CreateTypePlan, CreateViewPlan, DataSourceDesc, DropObjectsPlan, DropOwnedPlan,
-    HirRelationExpr, Index, MaterializedView, NetworkPolicyRule, NetworkPolicyRuleAction,
-    NetworkPolicyRuleDirection, OnHydration, Plan, PlanClusterOption, PlanNotice, PolicyAddress,
-    QueryContext, ReplicaConfig, Secret, Sink, Source, Table, TableDataSource, Type, VariableValue,
-    View, WebhookBodyFormat, WebhookHeaderFilters, WebhookHeaders, WebhookValidation, literal,
-    plan_utils, query, transform_ast,
+    CreateIndexPlan, CreateMaterializedViewPlan, CreateMetricSinkPlan, CreateNetworkPolicyPlan,
+    CreateRolePlan, CreateSchemaPlan, CreateSecretPlan, CreateSinkPlan, CreateSourcePlan,
+    CreateTablePlan, CreateTypePlan, CreateViewPlan, DataSourceDesc, DropObjectsPlan,
+    DropOwnedPlan, HirRelationExpr, Index, MaterializedView, MetricSink, NetworkPolicyRule,
+    NetworkPolicyRuleAction, NetworkPolicyRuleDirection, OnHydration, Plan, PlanClusterOption,
+    PlanNotice, PolicyAddress, QueryContext, ReplicaConfig, Secret, Sink, Source, Table,
+    TableDataSource, Type, VariableValue, View, WebhookBodyFormat, WebhookHeaderFilters,
+    WebhookHeaders, WebhookValidation, literal, plan_utils, query, transform_ast,
 };
 use crate::session::vars::{
     self, ENABLE_AUTO_SCALING_STRATEGY, ENABLE_CLUSTER_SCHEDULE_REFRESH,
     ENABLE_COLLECTION_PARTITION_BY, ENABLE_CREATE_TABLE_FROM_SOURCE, ENABLE_KAFKA_SINK_HEADERS,
-    ENABLE_REFRESH_EVERY_MVS, ENABLE_REPLICA_TARGETED_MATERIALIZED_VIEWS, VarInput,
+    ENABLE_METRIC_SINK, ENABLE_REFRESH_EVERY_MVS, ENABLE_REPLICA_TARGETED_MATERIALIZED_VIEWS,
+    VarInput,
 };
 use crate::{names, parse};
 
@@ -3296,7 +3297,7 @@ fn plan_sink(
                     });
                 }
             }
-            Sink | View | Index | Type | Func | Secret | Connection => {
+            Sink | MetricSink | View | Index | Type | Func | Secret | Connection => {
                 let name = scx.catalog.minimal_qualification(from.name());
                 return Err(PlanError::InvalidSinkFrom {
                     name: name.to_string(),
@@ -4206,6 +4207,122 @@ fn kafka_sink_builder(
     }))
 }
 
+pub fn describe_create_metric_sink(
+    _: &StatementContext,
+    _: CreateMetricSinkStatement<Aug>,
+) -> Result<StatementDesc, PlanError> {
+    Ok(StatementDesc::new(None))
+}
+
+/// Columns a metric sink source relation must expose, and the predicate its scalar type must
+/// satisfy. Order-independent. Columns may be nullable: null input is a runtime concern handled
+/// by the sink operator, not a plan-time rejection.
+const METRIC_SINK_SOURCE_COLUMNS: &[(&str, fn(&SqlScalarType) -> bool)] = &[
+    ("metric_name", |t| matches!(t, SqlScalarType::String)),
+    ("metric_type", |t| matches!(t, SqlScalarType::String)),
+    (
+        "labels",
+        |t| matches!(t, SqlScalarType::Map { value_type, .. } if matches!(**value_type, SqlScalarType::String)),
+    ),
+    ("value", |t| matches!(t, SqlScalarType::Float64)),
+    ("help", |t| matches!(t, SqlScalarType::String)),
+];
+
+fn validate_metric_sink_desc(desc: &RelationDesc) -> Result<(), PlanError> {
+    for (name, type_ok) in METRIC_SINK_SOURCE_COLUMNS {
+        let col = ColumnName::from(*name);
+        let (_, column_type) = desc
+            .get_by_name(&col)
+            .ok_or_else(|| sql_err!("metric sink source must expose column {:?}", name))?;
+        if !type_ok(&column_type.scalar_type) {
+            return Err(sql_err!(
+                "metric sink source column {:?} is not of the required type",
+                name
+            ));
+        }
+    }
+    Ok(())
+}
+
+/// Plans a metric sink over `stmt.from`, a relation exposing the columns in
+/// `METRIC_SINK_SOURCE_COLUMNS`.
+///
+/// The sink emits one Prometheus series per distinct `(metric_name, labels)` row, so a source
+/// with high-cardinality labels produces a correspondingly large series count. A null `value`
+/// does not publish as zero: it leaves that series absent from the exposition (a gap) until a
+/// non-null value arrives for the same `(metric_name, labels)`. Callers who want zero-fill
+/// instead of a gap write `coalesce(value, 0)` themselves.
+pub fn plan_create_metric_sink(
+    scx: &StatementContext,
+    mut stmt: CreateMetricSinkStatement<Aug>,
+) -> Result<Plan, PlanError> {
+    scx.require_feature_flag(&ENABLE_METRIC_SINK)?;
+
+    let CreateMetricSinkStatement {
+        name,
+        in_cluster,
+        if_not_exists,
+        from,
+    } = &mut stmt;
+
+    let Some(name) = name.clone() else {
+        return Err(PlanError::MissingName(CatalogItemType::MetricSink));
+    };
+    let if_not_exists = *if_not_exists;
+    let name = scx.allocate_qualified_name(normalize::unresolved_item_name(name)?)?;
+    let full_name = scx.catalog.resolve_full_name(&name);
+    let partial_name = PartialItemName::from(full_name.clone());
+    if let (false, Ok(item)) = (if_not_exists, scx.catalog.resolve_item(&partial_name)) {
+        return Err(PlanError::ItemAlreadyExists {
+            name: full_name.to_string(),
+            item_type: item.item_type(),
+        });
+    }
+
+    let from_item = scx.get_item_by_resolved_name(from)?;
+    {
+        use CatalogItemType::*;
+        match from_item.item_type() {
+            Table | Source | View | MaterializedView | Index => {}
+            Sink | MetricSink | Type | Func | Secret | Connection => {
+                sql_bail!(
+                    "cannot create metric sink from {} because it is a {}",
+                    scx.catalog.minimal_qualification(from_item.name()),
+                    from_item.item_type(),
+                );
+            }
+        }
+    }
+
+    let desc = from_item
+        .relation_desc()
+        .ok_or_else(|| sql_err!("item does not have a relation description"))?;
+    validate_metric_sink_desc(&desc)?;
+
+    let cluster_id = match in_cluster {
+        None => scx.resolve_cluster(None)?.id(),
+        Some(in_cluster) => in_cluster.id,
+    };
+    *in_cluster = Some(ResolvedClusterName {
+        id: cluster_id,
+        print_name: None,
+    });
+    let from_global_id = from_item.global_id();
+
+    let create_sql = normalize::create_statement(scx, Statement::CreateMetricSink(stmt))?;
+
+    Ok(Plan::CreateMetricSink(CreateMetricSinkPlan {
+        name,
+        metric_sink: MetricSink {
+            create_sql,
+            from: from_global_id,
+            cluster_id,
+        },
+        if_not_exists,
+        in_cluster: cluster_id,
+    }))
+}
+
 pub fn describe_create_index(
     _: &StatementContext,
     _: CreateIndexStatement<Aug>,
@@ -4239,7 +4356,7 @@ pub fn plan_create_index(
                     );
                 }
             }
-            Sink | Index | Type | Func | Secret | Connection => {
+            Sink | MetricSink | Index | Type | Func | Secret | Connection => {
                 sql_bail!(
                     "index cannot be created on {} because it is a {}",
                     on_name.full_name_str(),
@@ -5999,6 +6116,7 @@ fn dependency_prevents_drop(object_type: ObjectType, dep: &dyn CatalogItem) -> b
         | ObjectType::MaterializedView
         | ObjectType::Source
         | ObjectType::Sink
+        | ObjectType::MetricSink
         | ObjectType::Index
         | ObjectType::Role
         | ObjectType::Cluster
@@ -6015,6 +6133,7 @@ fn dependency_prevents_drop(object_type: ObjectType, dep: &dyn CatalogItem) -> b
             | CatalogItemType::View
             | CatalogItemType::MaterializedView
             | CatalogItemType::Sink
+            | CatalogItemType::MetricSink
             | CatalogItemType::Type
             | CatalogItemType::Secret
             | CatalogItemType::Connection => true,
@@ -6785,7 +6904,7 @@ pub fn plan_alter_item_set_cluster(
     // Prevent access to `SET CLUSTER` for unsupported objects.
     match object_type {
         ObjectType::MaterializedView => {}
-        ObjectType::Index | ObjectType::Sink | ObjectType::Source => {
+        ObjectType::Index | ObjectType::Sink | ObjectType::MetricSink | ObjectType::Source => {
             bail_unsupported!(29606, format!("ALTER {object_type} SET CLUSTER"))
         }
         ObjectType::Table
@@ -7233,6 +7352,7 @@ pub fn plan_alter_object_swap(
             | ObjectType::MaterializedView
             | ObjectType::Source
             | ObjectType::Sink
+            | ObjectType::MetricSink
             | ObjectType::Index
             | ObjectType::Type
             | ObjectType::Role
@@ -8322,6 +8442,7 @@ pub(crate) fn resolve_item_or_type<'a>(
         | ObjectType::MaterializedView
         | ObjectType::Source
         | ObjectType::Sink
+        | ObjectType::MetricSink
         | ObjectType::Index
         | ObjectType::Role
         | ObjectType::Cluster
diff --git a/src/sql/src/plan/statement/show.rs b/src/sql/src/plan/statement/show.rs
index 5e075d2f11a23..22313e9942e01 100644
--- a/src/sql/src/plan/statement/show.rs
+++ b/src/sql/src/plan/statement/show.rs
@@ -724,7 +724,8 @@ pub fn show_columns<'a>(
         | ty @ CatalogItemType::Func
         | ty @ CatalogItemType::Secret
         | ty @ CatalogItemType::Type
-        | ty @ CatalogItemType::Sink => {
+        | ty @ CatalogItemType::Sink
+        | ty @ CatalogItemType::MetricSink => {
             sql_bail!("{full_name} is a {ty} and so does not have columns");
         }
     }
diff --git a/src/sql/src/rbac.rs b/src/sql/src/rbac.rs
index 8b9c8cd826c91..0b4effc6c8467 100644
--- a/src/sql/src/rbac.rs
+++ b/src/sql/src/rbac.rs
@@ -719,6 +719,31 @@ fn generate_rbac_requirements(
                 ..Default::default()
             }
         }
+        Plan::CreateMetricSink(plan::CreateMetricSinkPlan {
+            name,
+            metric_sink,
+            if_not_exists: _,
+            in_cluster,
+        }) => {
+            let from_item = catalog.resolve_item_id(&metric_sink.from);
+            RbacRequirements {
+                ownership: vec![ObjectId::Item(from_item)],
+                privileges: vec![
+                    (
+                        SystemObjectId::Object(name.qualifiers.clone().into()),
+                        AclMode::CREATE,
+                        role_id,
+                    ),
+                    (
+                        SystemObjectId::Object((*in_cluster).into()),
+                        AclMode::CREATE,
+                        role_id,
+                    ),
+                ],
+                item_usage: &CREATE_ITEM_USAGE,
+                ..Default::default()
+            }
+        }
         Plan::CreateType(plan::CreateTypePlan { name, typ: _ }) => RbacRequirements {
             privileges: vec![(
                 SystemObjectId::Object(name.qualifiers.clone().into()),
@@ -1772,7 +1797,10 @@ fn generate_read_privileges_inner(
                 CatalogItemType::Type | CatalogItemType::Secret | CatalogItemType::Connection => {
                     privileges.push((SystemObjectId::Object(id.into()), AclMode::USAGE, role_id));
                 }
-                CatalogItemType::Sink | CatalogItemType::Index | CatalogItemType::Func => {}
+                CatalogItemType::Sink
+                | CatalogItemType::MetricSink
+                | CatalogItemType::Index
+                | CatalogItemType::Func => {}
             }
         }
     }
@@ -1886,6 +1914,7 @@ pub const fn all_object_privileges(object_type: SystemObjectType) -> AclMode {
         SystemObjectType::Object(ObjectType::MaterializedView) => AclMode::SELECT,
         SystemObjectType::Object(ObjectType::Source) => AclMode::SELECT,
         SystemObjectType::Object(ObjectType::Sink) => EMPTY_ACL_MODE,
+        SystemObjectType::Object(ObjectType::MetricSink) => EMPTY_ACL_MODE,
         SystemObjectType::Object(ObjectType::Index) => EMPTY_ACL_MODE,
         SystemObjectType::Object(ObjectType::Type) => AclMode::USAGE,
         SystemObjectType::Object(ObjectType::Role) => EMPTY_ACL_MODE,
@@ -1917,6 +1946,7 @@ const fn default_builtin_object_acl_mode(object_type: ObjectType) -> AclMode {
         | ObjectType::Source => AclMode::SELECT,
         ObjectType::Type | ObjectType::Schema => AclMode::USAGE,
         ObjectType::Sink
+        | ObjectType::MetricSink
         | ObjectType::Index
         | ObjectType::Role
         | ObjectType::Cluster
diff --git a/src/sql/src/session/vars/definitions.rs b/src/sql/src/session/vars/definitions.rs
index d8c2c2a0bdf6f..28882a3a94a98 100644
--- a/src/sql/src/session/vars/definitions.rs
+++ b/src/sql/src/session/vars/definitions.rs
@@ -2192,6 +2192,16 @@ feature_flags!(
         default: false,
         enable_for_item_parsing: true,
     },
+    {
+        name: enable_metric_sink,
+        desc: "CREATE METRIC SINK",
+        default: false,
+        // Metric sink catalog items are durably persisted like any other item (their type is
+        // re-derived by re-parsing `create_sql` on boot, see `item_type` in durable/objects.rs),
+        // so force the flag on for item parsing. Otherwise boot would fail to re-parse a metric
+        // sink created while the flag was enabled.
+        enable_for_item_parsing: true,
+    },
     {
         name: enable_unlimited_retain_history,
         desc: "Disable limits on RETAIN HISTORY (below 1s default, and 0 disables compaction).",
diff --git a/test/testdrive/metric-sink.td b/test/testdrive/metric-sink.td
new file mode 100644
index 0000000000000..6f5229c530e91
--- /dev/null
+++ b/test/testdrive/metric-sink.td
@@ -0,0 +1,154 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# End-to-end test for CREATE METRIC SINK: verifies that the series computed by
+# a metric sink's dataflow actually reach the cluster replica's Prometheus
+# registry, as observed through `mz_introspection.mz_cluster_prometheus_metrics`
+# (the same registry the sink's collector registers into, scraped on
+# `compute_prometheus_introspection_scrape_interval`), and that DROP METRIC
+# SINK removes both the user series and the sink's companion gauges.
+#
+# This test does not depend on the deferred `mz_metric_sinks` catalog view; it
+# reads the sink's id from the `sink` label the operator stamps on its own
+# companion gauges.
+
+$ set-sql-timeout duration=60s
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM SET enable_metric_sink = true;
+
+> CREATE TABLE t (k text NOT NULL, val double)
+
+> INSERT INTO t VALUES ('a', 1), ('b', 2)
+
+> CREATE VIEW v AS
+  SELECT 'mz_metric_sink_test_value'::text AS metric_name,
+         'gauge'::text AS metric_type,
+         map_build(LIST[ROW('k', k)])::map[text=>text] AS labels,
+         val AS value,
+         'a metric emitted by the metric sink e2e test'::text AS help
+  FROM t
+
+# The FROM relation must expose the canonical metric-sink columns. A relation
+# missing `value` (or with a wrongly-typed column) is rejected at plan time.
+> CREATE VIEW bad_missing_value AS
+  SELECT 'm'::text AS metric_name,
+         'gauge'::text AS metric_type,
+         map_build(LIST[ROW('k', k)])::map[text=>text] AS labels,
+         'help'::text AS help
+  FROM t
+
+! CREATE METRIC SINK bad IN CLUSTER quickstart FROM bad_missing_value
+contains:metric sink source must expose column "value"
+
+> CREATE VIEW bad_wrong_type AS
+  SELECT 'm'::text AS metric_name,
+         'gauge'::text AS metric_type,
+         map_build(LIST[ROW('k', k)])::map[text=>text] AS labels,
+         val::int AS value,
+         'help'::text AS help
+  FROM t
+
+! CREATE METRIC SINK bad IN CLUSTER quickstart FROM bad_wrong_type
+contains:metric sink source column "value" is not of the required type
+
+> DROP VIEW bad_missing_value
+
+> DROP VIEW bad_wrong_type
+
+> CREATE METRIC SINK s IN CLUSTER quickstart FROM v
+
+# One family, two series: one per distinct label set.
+> SELECT count(*) FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_test_value'
+2
+
+> SELECT value = 1 FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_test_value' AND labels -> 'k' = 'a'
+true
+
+> SELECT value = 2 FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_test_value' AND labels -> 'k' = 'b'
+true
+
+> SELECT count(*) FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_test_value'
+    AND metric_type = 'gauge'
+    AND help = 'a metric emitted by the metric sink e2e test'
+2
+
+# Read the sink's id off the companion gauge it stamps with a `sink` label.
+$ set-from-sql var=sink-id
+SELECT labels -> 'sink' FROM mz_introspection.mz_cluster_prometheus_metrics
+WHERE metric_name = 'mz_metric_sink_frontier_ms'
+
+# The companion gauges exist, are labeled with this sink's id, and read healthy
+# (no errors, no skipped/conflicting/colliding series).
+> SELECT value > 0 FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_frontier_ms' AND labels -> 'sink' = '${sink-id}'
+true
+
+> SELECT value = 0 FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_errors' AND labels -> 'sink' = '${sink-id}'
+true
+
+> SELECT value = 0 FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_skipped' AND labels -> 'sink' = '${sink-id}'
+true
+
+# A null `value` does not publish a zero: the `(metric_name, labels)` series
+# for 'c' is absent from the registry (a gap) while its only live value is
+# null, and the sink counts it in `mz_metric_sink_null_values`.
+> INSERT INTO t VALUES ('c', NULL)
+
+> SELECT count(*) FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_test_value'
+2
+
+> SELECT count(*) FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_test_value' AND labels -> 'k' = 'c'
+0
+
+> SELECT value = 1 FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_null_values' AND labels -> 'sink' = '${sink-id}'
+true
+
+# The gap closes once a non-null value arrives for the same series.
+> UPDATE t SET val = 3 WHERE k = 'c'
+
+> SELECT count(*) FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_test_value'
+3
+
+> SELECT value = 3 FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_test_value' AND labels -> 'k' = 'c'
+true
+
+> SELECT value = 0 FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_null_values' AND labels -> 'sink' = '${sink-id}'
+true
+
+> DROP METRIC SINK s
+
+# The user series and all of this sink's companion gauges are retracted from
+# the registry once the next scrape observes the collector is gone.
+> SELECT count(*) FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE metric_name = 'mz_metric_sink_test_value'
+0
+
+> SELECT count(*) FROM mz_introspection.mz_cluster_prometheus_metrics
+  WHERE labels -> 'sink' = '${sink-id}'
+0
+
+> DROP VIEW v
+
+> DROP TABLE t
+
+$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+ALTER SYSTEM RESET enable_metric_sink;