metric sinks: coordinator per-replica install scaffold (SQL-555)

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

@@ -700,6 +700,8 @@ The view is defined as the transitive closure of [`mz_object_dependencies`](#mz_
 | `object_id`             | [`text`]     | The ID of the dependent object. Corresponds to [`mz_objects.id`](../mz_catalog/#mz_objects).                          |
 | `referenced_object_id`  | [`text`]     | The ID of the (possibly transitively) referenced object. Corresponds to [`mz_objects.id`](../mz_catalog/#mz_objects). |
 
+<!-- RELATION_SPEC_UNDOCUMENTED mz_internal.mz_metric_sinks -->
+
 ## `mz_notices`
 
 {{< public-preview />}}

misc/python/materialize/checks/all_checks/metric_sink.py

@@ -0,0 +1,92 @@
+# 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.
+from textwrap import dedent
+
+from materialize.checks.actions import Testdrive
+from materialize.checks.checks import Check
+from materialize.checks.executors import Executor
+from materialize.mz_version import MzVersion
+
+
+class MetricSink(Check):
+    """Boot has to re-parse a metric sink's `create_sql` back into a catalog
+    item, rebuild the dependency edge to the relation it reads, and re-render
+    the sink's dataflow. A sink whose dataflow does not come back publishes no
+    metrics, so the check probes for the dataflow, not just the item."""
+
+    def _can_run(self, e: Executor) -> bool:
+        return self.base_version >= MzVersion.parse_mz("v26.36.0-dev")
+
+    def initialize(self) -> Testdrive:
+        return Testdrive(dedent("""
+                > CREATE TABLE metric_sink_table (metric_name text, metric_type text, labels map[text=>text], value double, help text)
+
+                > INSERT INTO metric_sink_table VALUES ('a', 'gauge', '{x=>y}', 1, 'help a')
+
+                > CREATE VIEW metric_sink_view AS SELECT * FROM metric_sink_table
+
+                > CREATE METRIC SINK metric_sink_one IN CLUSTER quickstart FROM metric_sink_view
+                """))
+
+    def manipulate(self) -> list[Testdrive]:
+        return [
+            Testdrive(dedent(s))
+            for s in [
+                """
+                > INSERT INTO metric_sink_table VALUES ('b', 'counter', '{x=>z}', 2, 'help b')
+
+                > CREATE METRIC SINK metric_sink_two IN CLUSTER quickstart FROM metric_sink_view
+                """,
+                """
+                > INSERT INTO metric_sink_table VALUES ('c', 'gauge', '{}', 3, 'help c')
+
+                > CREATE METRIC SINK IF NOT EXISTS metric_sink_three IN CLUSTER quickstart FROM metric_sink_view
+                """,
+            ]
+        ]
+
+    def validate(self) -> Testdrive:
+        return Testdrive(dedent("""
+                > SHOW METRIC SINKS
+                metric_sink_one   metric_sink_view quickstart
+                metric_sink_three metric_sink_view quickstart
+                metric_sink_two   metric_sink_view quickstart
+
+                # The FROM edge came back too, so the view is still pinned.
+                ! DROP VIEW metric_sink_view
+                contains:still depended upon by metric sink
+
+                > SELECT count(*) FROM metric_sink_view
+                3
+
+                # Each re-rendered dataflow registers a collector that stamps
+                # its own sink id on a frontier gauge, so three distinct ids
+                # means all three dataflows came back. The registry is only
+                # sampled every `compute_prometheus_introspection_scrape_interval`,
+                # so give the scrape room to land.
+                $ set-sql-timeout duration=60s
+
+                # Every replica has its own registry, so this introspection
+                # relation can only be read with a replica targeted.
+                $ set-from-sql var=replica-name
+                SELECT r.name
+                FROM mz_catalog.mz_cluster_replicas r
+                JOIN mz_catalog.mz_clusters c ON c.id = r.cluster_id
+                WHERE c.name = 'quickstart'
+                ORDER BY r.name
+                LIMIT 1
+
+                > SET cluster_replica = ${replica-name}
+
+                > SELECT count(DISTINCT labels -> 'sink') FROM mz_introspection.mz_cluster_prometheus_metrics
+                  WHERE metric_name = 'mz_metric_sink_frontier_ms'
+                3
+
+                > RESET cluster_replica
+                """))

misc/python/materialize/mzcompose/__init__.py

@@ -90,6 +90,7 @@ def get_minimal_system_parameters(
         "enable_lgalloc": "false",
         "enable_load_generator_counter": "true",
         "enable_logical_compaction_window": "true",
+        "enable_metric_sink": "true",
         "enable_multi_worker_storage_persist_sink": "true",
         "enable_multi_replica_sources": "true",
         "enable_rbac_checks": "true",

src/adapter/src/catalog.rs

@@ -1694,6 +1694,7 @@ pub(crate) fn comment_id_to_audit_object_type(id: CommentObjectId) -> 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,
@@ -1724,6 +1725,7 @@ 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,
+            mz_sql::catalog::ObjectType::MetricSink => ObjectType::MetricSink,
             mz_sql::catalog::ObjectType::Index => ObjectType::Index,
             mz_sql::catalog::ObjectType::Type => ObjectType::Type,
             mz_sql::catalog::ObjectType::Role => ObjectType::Role,

src/adapter/src/catalog/apply.rs

@@ -1558,7 +1558,7 @@ impl CatalogState {
     /// Set the optimized plan for the item identified by `id`.
     ///
     /// # Panics
-    /// If the item is not an `Index` or `MaterializedView`.
+    /// If the item is not an `Index`, `MaterializedView`, or `MetricSink`.
     pub(super) fn set_optimized_plan(
         &mut self,
         id: GlobalId,
@@ -1569,14 +1569,15 @@ 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()),
         }
     }
 
     /// Set the physical plan for the item identified by `id`.
     ///
     /// # Panics
-    /// If the item is not an `Index` or `MaterializedView`.
+    /// If the item is not an `Index`, `MaterializedView`, or `MetricSink`.
     pub(super) fn set_physical_plan(
         &mut self,
         id: GlobalId,
@@ -1587,14 +1588,15 @@ 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()),
         }
     }
 
     /// Set the `DataflowMetainfo` for the item identified by `id`.
     ///
     /// # Panics
-    /// If the item is not an `Index` or `MaterializedView`.
+    /// If the item is not an `Index`, `MaterializedView`, or `MetricSink`.
     pub(super) fn set_dataflow_metainfo(
         &mut self,
         id: GlobalId,
@@ -1622,6 +1624,7 @@ impl CatalogState {
         match entry.item_mut() {
             CatalogItem::Index(idx) => idx.dataflow_metainfo = Some(metainfo),
             CatalogItem::MaterializedView(mv) => mv.dataflow_metainfo = Some(metainfo),
+            CatalogItem::MetricSink(ms) => ms.dataflow_metainfo = Some(metainfo),
             other => panic!("set_dataflow_metainfo called on {} ({:?})", id, other.typ()),
         }
     }
@@ -2443,7 +2446,8 @@ fn sort_updates(updates: Vec<StateUpdate>) -> Vec<StateUpdate> {
                 CatalogItemType::Table => tables.push(update),
                 CatalogItemType::View
                 | CatalogItemType::MaterializedView
-                | CatalogItemType::Index => derived_items.push(update),
+                | CatalogItemType::Index
+                | CatalogItemType::MetricSink => derived_items.push(update),
                 CatalogItemType::Sink => sinks.push(update),
             }
         }
@@ -2508,7 +2512,8 @@ fn sort_updates(updates: Vec<StateUpdate>) -> Vec<StateUpdate> {
                 CatalogItemType::Table => tables.push(update),
                 CatalogItemType::View
                 | CatalogItemType::MaterializedView
-                | CatalogItemType::Index => derived_items.push(update),
+                | CatalogItemType::Index
+                | CatalogItemType::MetricSink => derived_items.push(update),
                 CatalogItemType::Sink => sinks.push(update),
             }
         }

src/adapter/src/catalog/builtin_table_updates.rs

@@ -201,7 +201,9 @@ impl CatalogState {
             CatalogItem::Func(func) => {
                 self.pack_func_update(id, schema_id, name, owner_id, func, diff)
             }
-            CatalogItem::Log(_) | CatalogItem::Secret(_) => vec![],
+            // `mz_metric_sinks` is a materialized view derived from
+            // `mz_catalog_raw`, so metric sinks emit no builtin-table row here.
+            CatalogItem::Log(_) | CatalogItem::Secret(_) | CatalogItem::MetricSink(_) => vec![],
             // Connection details (mz_kafka_connections, mz_ssh_tunnel_connections,
             // mz_aws_connections, mz_aws_privatelink_connections) are now derived
             // from the persisted create_sql by materialized views over

src/adapter/src/catalog/consistency.rs

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

src/adapter/src/catalog/open.rs

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

src/adapter/src/catalog/open/builtin_schema_migration.rs

@@ -355,6 +355,17 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
             MZ_CATALOG_SCHEMA,
             "mz_aws_privatelink_connections",
         ),
+        // `mz_audit_events` learned the `metric-sink` object type. Its `object_type`
+        // column carries `ASSERT NOT NULL`, so without the new CASE arm a single
+        // metric-sink audit event errors the whole collection. See the NOTE above:
+        // this version must stay at the workspace's current dev version until the
+        // change ships.
+        MigrationStep::replacement(
+            "26.38.0-dev.0",
+            CatalogItemType::MaterializedView,
+            MZ_CATALOG_SCHEMA,
+            "mz_audit_events",
+        ),
     ]
 });
 

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;
@@ -459,6 +459,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) {
@@ -1504,6 +1510,18 @@ impl CatalogState {
                 physical_plan: None,
                 dataflow_metainfo: None,
             }),
+            Plan::CreateMetricSink(CreateMetricSinkPlan { metric_sink, .. }) => {
+                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,
+                })
+            }
             Plan::CreateSink(CreateSinkPlan {
                 sink,
                 with_snapshot,
@@ -1924,6 +1942,7 @@ impl CatalogState {
             CatalogItemType::Table
             | CatalogItemType::Source
             | CatalogItemType::Sink
+            | CatalogItemType::MetricSink
             | CatalogItemType::View
             | CatalogItemType::MaterializedView
             | CatalogItemType::Index
@@ -2746,6 +2765,7 @@ impl CatalogState {
             | CommentObjectId::MaterializedView(id)
             | CommentObjectId::Source(id)
             | CommentObjectId::Sink(id)
+            | CommentObjectId::MetricSink(id)
             | CommentObjectId::Index(id)
             | CommentObjectId::Func(id)
             | CommentObjectId::Connection(id)
@@ -2775,6 +2795,7 @@ impl CatalogState {
             | CommentObjectId::MaterializedView(id)
             | CommentObjectId::Source(id)
             | CommentObjectId::Sink(id)
+            | CommentObjectId::MetricSink(id)
             | CommentObjectId::Index(id)
             | CommentObjectId::Func(id)
             | CommentObjectId::Connection(id)

src/adapter/src/catalog/timeline.rs

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

src/adapter/src/catalog/transact.rs

@@ -906,7 +906,8 @@ impl Catalog {
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)
-                    | CatalogItem::Connection(_) => {}
+                    | CatalogItem::Connection(_)
+                    | CatalogItem::MetricSink(_) => {}
                 }
             }
         }
@@ -1718,7 +1719,10 @@ impl Catalog {
                     | CatalogItem::Type(_)
                     | CatalogItem::Func(_)
                     | CatalogItem::Secret(_)
-                    | CatalogItem::Connection(_) => (),
+                    | CatalogItem::Connection(_)
+                    // Metric sinks write to the replica's metrics registry, never to persist,
+                    // so there is no storage collection to create.
+                    | CatalogItem::MetricSink(_) => (),
                 }
 
                 let system_user = session.map_or(false, |s| s.user().is_system_user());
@@ -1902,7 +1906,8 @@ impl Catalog {
                         | CatalogItem::Type(_)
                         | CatalogItem::Func(_)
                         | CatalogItem::Secret(_)
-                        | CatalogItem::Connection(_) => EventDetails::IdFullNameV1(IdFullNameV1 {
+                        | CatalogItem::Connection(_)
+                        | CatalogItem::MetricSink(_) => EventDetails::IdFullNameV1(IdFullNameV1 {
                             id: id.to_string(),
                             name,
                         }),