metric sinks: SHOW, catalog visibility, and RBAC (SQL-572)

Problem:

A metric sink was invisible, `mz_objects` did not carry it, and there
was no way to read back the SQL that created it.

Solution:

`SHOW METRIC SINKS [FROM <schema>] [IN CLUSTER <c>] [LIKE ...]` lists
name, `from`, and cluster. `SHOW [REDACTED] CREATE METRIC SINK` replays
`create_sql`. Both require `enable_metric_sink`.

`mz_internal.mz_metric_sinks` exposes what the catalog knows about a
sink, shaped like `mz_catalog.mz_sinks`, and joins into `mz_objects` as
type `metric-sink`. It is a materialized view derived from
`mz_catalog_raw`. It carries `owner_id`, so a sink's owner is now
visible in the catalog and its ownership checks are testable. Like
`mz_sinks`, it is indexed on `id` so the `mz_objects` union reuses the
arrangement.

The relation carries only the columns something reads. `create_sql` and
`redacted_create_sql` have no reader, and `SHOW CREATE` is how you get
the SQL back.

`enable_metric_sink` gates the DDL and the SHOW verbs, not the catalog
relation. `mz_metric_sinks` is always present and public, so with the
flag off it is simply empty.

Testing:

- `metric_sink.slt` covers discovery and access control: the
`mz_metric_sinks` row resolving to its `FROM` relation, cluster, schema,
and owner, `mz_objects` and `mz_show_all_objects` membership, the three
`SHOW METRIC SINKS` filters, the `SHOW CREATE` round-trip (plain and
redacted), the audit rows, and privileges.

- The restart platform check reads `SHOW METRIC SINKS` instead of
probing with a `CREATE` expected to fail.

Co-authored-by: Moritz Hoffmann <antiguru@gmail.com>

Author: mtabebe

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

@@ -716,6 +716,12 @@ 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 -->
+<!-- TODO(metric-sink): promote to a documented RELATION_SPEC once
+     `enable_metric_sink` defaults on. The relation already ships full
+     per-column comments, so this "undocumented" marker is only correct while
+     the feature is gated off. -->
+
 ## `mz_notices`
 
 {{< public-preview />}}

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

@@ -61,24 +61,13 @@ def manipulate(self) -> list[Testdrive]:
         ]
 
     def validate(self) -> Testdrive:
-        # Recreating a sink that is already there is how we probe for it without
-        # mutating anything: metric sinks are not in `mz_objects` yet, and
-        # `mz_catalog_raw` needs a system connection.
-        #
-        # TODO(SQL-572): once metric sinks join `mz_objects` (and the
-        # `mz_metric_sinks` builtin view), probe survival by SELECTing the sink
-        # row instead of re-issuing CREATE and matching "already exists". The
-        # current probe is a proxy: it confirms the catalog item was re-parsed on
-        # boot without needing a system connection.
         return Testdrive(dedent("""
-                ! CREATE METRIC SINK metric_sink_schema_renamed.metric_sink_one FROM metric_sink_view_renamed WITH (PREFIX = 'mz_metric_sink_one_')
-                contains:metric sink "materialize.metric_sink_schema_renamed.metric_sink_one" already exists
+                > SHOW METRIC SINKS
+                metric_sink_three metric_sink_view_renamed quickstart
+                metric_sink_two   metric_sink_view_renamed quickstart
 
-                ! CREATE METRIC SINK metric_sink_two FROM metric_sink_view_renamed WITH (PREFIX = 'mz_metric_sink_two_')
-                contains:metric sink "materialize.public.metric_sink_two" already exists
-
-                ! CREATE METRIC SINK metric_sink_three FROM metric_sink_view_renamed WITH (PREFIX = 'mz_metric_sink_three_')
-                contains:metric sink "materialize.public.metric_sink_three" already exists
+                > SHOW METRIC SINKS FROM metric_sink_schema_renamed
+                metric_sink_one metric_sink_view_renamed quickstart
 
                 # The FROM edge came back too, so the view is still pinned, under the
                 # name the rename gave it.

misc/wasm/src/sql-parser-wasm/src/lib.rs

@@ -100,6 +100,7 @@ fn refine_show_kind(stmt: &ShowStatement<Raw>) -> &'static str {
         ShowStatement::ShowCreateSource(_) => "show_create_source",
         ShowStatement::ShowCreateTable(_) => "show_create_table",
         ShowStatement::ShowCreateSink(_) => "show_create_sink",
+        ShowStatement::ShowCreateMetricSink(_) => "show_create_metric_sink",
         ShowStatement::ShowCreateIndex(_) => "show_create_index",
         ShowStatement::ShowCreateConnection(_) => "show_create_connection",
         ShowStatement::ShowCreateCluster(_) => "show_create_cluster",

src/adapter/src/catalog/builtin_table_updates.rs

@@ -190,14 +190,9 @@ impl CatalogState {
             CatalogItem::Func(func) => {
                 self.pack_func_update(id, schema_id, name, owner_id, func, diff)
             }
-            // A metric sink packs no builtin-table row, so it holds a catalog name that no
-            // catalog relation reports. SQL-572 adds the `mz_metric_sinks` view. Until then,
-            // listing or dropping a metric sink requires knowing its name out of band.
-            //
-            // NOTE: creating a metric sink takes SELECT on the FROM relation, not ownership
-            // of it, so once metric sinks are user-creatable this gap would let a reader
-            // egress another role's rows with no catalog relation the owner could see it in.
-            // SQL-572 has to land before user-facing metric sink DDL does.
+            // Tables, views, and metric sinks are exposed through materialized
+            // views derived from `mz_catalog_raw`, and logs and secrets never
+            // had builtin-table rows, so none pack a row here.
             CatalogItem::Table(_)
             | CatalogItem::View(_)
             | CatalogItem::Log(_)

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

@@ -390,9 +390,10 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
             MZ_CATALOG_SCHEMA,
             "mz_audit_events",
         ),
-        // Required because we added the `mz_object_graph_edges_ind` builtin index.
-        // make_mz_indexes inlines the builtin-index set as VALUES, so any add or
-        // remove changes its SQL fingerprint and requires an explicit replacement.
+        // Required because we added the `mz_object_graph_edges_ind` and
+        // `mz_metric_sinks_ind` builtin indexes. make_mz_indexes inlines the
+        // builtin-index set as VALUES, so any add or remove changes its SQL
+        // fingerprint and requires an explicit replacement.
         MigrationStep::replacement(
             "26.39.0-dev.0",
             CatalogItemType::MaterializedView,

src/catalog/src/builtin.rs

@@ -1172,6 +1172,7 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::MaterializedView(&MZ_SYSTEM_PRIVILEGES),
         Builtin::MaterializedView(&MZ_COMMENTS),
         Builtin::Table(&MZ_WEBHOOKS_SOURCES),
+        Builtin::MaterializedView(&MZ_METRIC_SINKS),
         Builtin::Table(&MZ_HISTORY_RETENTION_STRATEGIES),
         Builtin::MaterializedView(&MZ_MATERIALIZED_VIEWS),
         Builtin::Table(&MZ_MATERIALIZED_VIEW_REFRESH_STRATEGIES),
@@ -1419,6 +1420,7 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::Index(&MZ_ROLES_IND),
         Builtin::Index(&MZ_SOURCES_IND),
         Builtin::Index(&MZ_SINKS_IND),
+        Builtin::Index(&MZ_METRIC_SINKS_IND),
         Builtin::Index(&MZ_MATERIALIZED_VIEWS_IND),
         Builtin::Index(&MZ_SOURCE_STATUSES_IND),
         Builtin::Index(&MZ_SOURCE_STATUS_HISTORY_IND),

src/catalog/src/builtin/mz_catalog.rs

@@ -3455,7 +3455,7 @@ pub static MZ_OBJECTS: LazyLock<BuiltinView> = LazyLock::new(|| {
             ("oid", "A PostgreSQL-compatible OID for the object."),
             ("schema_id", "The ID of the schema to which the object belongs. Corresponds to `mz_schemas.id`."),
             ("name", "The name of the object."),
-            ("type", "The type of the object: one of `table`, `source`, `view`, `materialized-view`, `sink`, `index`, `connection`, `secret`, `type`, or `function`."),
+            ("type", "The type of the object: one of `table`, `source`, `view`, `materialized-view`, `sink`, `metric-sink`, `index`, `connection`, `secret`, `type`, or `function`."),
             ("owner_id", "The role ID of the owner of the object. Corresponds to `mz_roles.id`."),
             ("cluster_id", "The ID of the cluster maintaining the source, materialized view, index, or sink. Corresponds to `mz_clusters.id`. `NULL` for other object types."),
             ("privileges", "The privileges belonging to the object."),
@@ -3464,6 +3464,8 @@ pub static MZ_OBJECTS: LazyLock<BuiltinView> = LazyLock::new(|| {
         "SELECT id, oid, schema_id, name, type, owner_id, cluster_id, privileges FROM mz_catalog.mz_relations
 UNION ALL
     SELECT id, oid, schema_id, name, 'sink', owner_id, cluster_id, NULL::mz_catalog.mz_aclitem[] FROM mz_catalog.mz_sinks
+UNION ALL
+    SELECT id, oid, schema_id, name, 'metric-sink', owner_id, cluster_id, NULL::mz_catalog.mz_aclitem[] FROM mz_internal.mz_metric_sinks
 UNION ALL
     SELECT mz_indexes.id, mz_indexes.oid, mz_relations.schema_id, mz_indexes.name, 'index', mz_indexes.owner_id, mz_indexes.cluster_id, NULL::mz_catalog.mz_aclitem[]
     FROM mz_catalog.mz_indexes

src/catalog/src/builtin/mz_internal.rs

@@ -3689,6 +3689,122 @@ pub static MZ_WEBHOOKS_SOURCES: LazyLock<BuiltinTable> = LazyLock::new(|| Builti
     }),
 });
 
+pub static MZ_METRIC_SINKS: LazyLock<BuiltinMaterializedView> = LazyLock::new(|| {
+    BuiltinMaterializedView {
+        name: "mz_metric_sinks",
+        schema: MZ_INTERNAL_SCHEMA,
+        oid: oid::MV_MZ_METRIC_SINKS_OID,
+        desc: RelationDesc::builder()
+            .with_column("id", SqlScalarType::String.nullable(false))
+            .with_column("oid", SqlScalarType::Oid.nullable(false))
+            .with_column("schema_id", SqlScalarType::String.nullable(false))
+            .with_column("name", SqlScalarType::String.nullable(false))
+            .with_column("from_id", SqlScalarType::String.nullable(false))
+            .with_column("cluster_id", SqlScalarType::String.nullable(false))
+            .with_column("owner_id", SqlScalarType::String.nullable(false))
+            .with_key(vec![0])
+            .with_key(vec![1])
+            .finish(),
+        column_comments: BTreeMap::from_iter([
+            ("id", "Materialize's unique ID for the metric sink."),
+            ("oid", "A PostgreSQL-compatible OID for the metric sink."),
+            (
+                "schema_id",
+                "The ID of the schema to which the metric sink belongs. Corresponds to `mz_schemas.id`.",
+            ),
+            ("name", "The name of the metric sink."),
+            (
+                "from_id",
+                "The ID of the relation the metric sink reads. Corresponds to `mz_objects.id`.",
+            ),
+            (
+                "cluster_id",
+                "The ID of the cluster maintaining the metric sink. Corresponds to `mz_clusters.id`.",
+            ),
+            (
+                "owner_id",
+                "The role ID of the owner of the metric sink. Corresponds to `mz_roles.id`.",
+            ),
+        ]),
+        sql: "
+IN CLUSTER mz_catalog_server
+WITH (
+    ASSERT NOT NULL id,
+    ASSERT NOT NULL oid,
+    ASSERT NOT NULL schema_id,
+    ASSERT NOT NULL name,
+    ASSERT NOT NULL from_id,
+    ASSERT NOT NULL cluster_id,
+    ASSERT NOT NULL owner_id
+) AS
+SELECT
+    mz_internal.parse_catalog_id(data->'key'->'gid') AS id,
+    (data->'value'->>'oid')::oid AS oid,
+    mz_internal.parse_catalog_id(data->'value'->'schema_id') AS schema_id,
+    data->'value'->>'name' AS name,
+    parsed->>'from_id' AS from_id,
+    parsed->>'cluster_id' AS cluster_id,
+    mz_internal.parse_catalog_id(data->'value'->'owner_id') AS owner_id
+FROM
+    mz_internal.mz_catalog_raw
+    CROSS JOIN LATERAL (
+        SELECT mz_internal.parse_catalog_create_sql(data->'value'->'definition'->'V1'->>'create_sql')
+    ) AS l(parsed)
+WHERE
+    data->>'kind' = 'Item' AND
+    parsed->>'type' = 'metric-sink'",
+        is_retained_metrics_object: false,
+        access: vec![PUBLIC_SELECT],
+        ontology: Some(Ontology {
+            entity_name: "metric-sink",
+            description: "A sink that exports metrics about a relation",
+            links: &const {
+                [
+                    OntologyLink {
+                        name: "in_schema",
+                        target: "schema",
+                        properties: LinkProperties::fk("schema_id", "id", Cardinality::ManyToOne),
+                    },
+                    OntologyLink {
+                        name: "reads_relation",
+                        target: "relation",
+                        properties: LinkProperties::fk("from_id", "id", Cardinality::ManyToOne),
+                    },
+                    OntologyLink {
+                        name: "runs_on_cluster",
+                        target: "cluster",
+                        properties: LinkProperties::fk("cluster_id", "id", Cardinality::ManyToOne),
+                    },
+                    OntologyLink {
+                        name: "owned_by",
+                        target: "role",
+                        properties: LinkProperties::fk("owner_id", "id", Cardinality::ManyToOne),
+                    },
+                ]
+            },
+            column_semantic_types: &const {
+                [
+                    ("id", SemanticType::CatalogItemId),
+                    ("oid", SemanticType::OID),
+                    ("schema_id", SemanticType::SchemaId),
+                    ("from_id", SemanticType::CatalogItemId),
+                    ("cluster_id", SemanticType::ClusterId),
+                    ("owner_id", SemanticType::RoleId),
+                ]
+            },
+        }),
+    }
+});
+
+pub const MZ_METRIC_SINKS_IND: BuiltinIndex = BuiltinIndex {
+    name: "mz_metric_sinks_ind",
+    schema: MZ_INTERNAL_SCHEMA,
+    oid: oid::INDEX_MZ_METRIC_SINKS_IND_OID,
+    sql: "IN CLUSTER mz_catalog_server
+ON mz_internal.mz_metric_sinks (id)",
+    is_retained_metrics_object: false,
+};
+
 pub static MZ_HISTORY_RETENTION_STRATEGIES: LazyLock<BuiltinTable> = LazyLock::new(|| {
     BuiltinTable {
         name: "mz_history_retention_strategies",
@@ -3887,6 +4003,7 @@ pub static MZ_OBJECTS_ID_NAMESPACE_TYPES: LazyLock<BuiltinView> = LazyLock::new(
             ('materialized-view'),
             ('source'),
             ('sink'),
+            ('metric-sink'),
             ('index'),
             ('connection'),
             ('type'),

src/pgrepr-consts/src/oid.rs

@@ -826,3 +826,5 @@ pub const VIEW_MZ_OBJECT_GRAPH_EDGES_OID: u32 = 17116;
 pub const INDEX_MZ_OBJECT_GRAPH_EDGES_IND_OID: u32 = 17117;
 pub const VIEW_MZ_BUILTIN_TABLES_OID: u32 = 17118;
 pub const VIEW_MZ_BUILTIN_VIEWS_OID: u32 = 17119;
+pub const MV_MZ_METRIC_SINKS_OID: u32 = 17120;
+pub const INDEX_MZ_METRIC_SINKS_IND_OID: u32 = 17121;

src/sql-parser/src/ast/defs/statement.rs

@@ -3589,6 +3589,9 @@ pub enum ShowObjectType<T: AstInfo> {
     Sink {
         in_cluster: Option<T::ClusterName>,
     },
+    MetricSink {
+        in_cluster: Option<T::ClusterName>,
+    },
     Type,
     Role,
     Cluster,
@@ -3641,6 +3644,7 @@ impl<T: AstInfo> AstDisplay for ShowObjectsStatement<T> {
             ShowObjectType::View => "VIEWS",
             ShowObjectType::Source { .. } => "SOURCES",
             ShowObjectType::Sink { .. } => "SINKS",
+            ShowObjectType::MetricSink { .. } => "METRIC SINKS",
             ShowObjectType::Type => "TYPES",
             ShowObjectType::Role => "ROLES",
             ShowObjectType::Cluster => "CLUSTERS",
@@ -3681,6 +3685,7 @@ impl<T: AstInfo> AstDisplay for ShowObjectsStatement<T> {
             ShowObjectType::MaterializedView { in_cluster }
             | ShowObjectType::Index { in_cluster, .. }
             | ShowObjectType::Sink { in_cluster }
+            | ShowObjectType::MetricSink { in_cluster }
             | ShowObjectType::Source { in_cluster } => {
                 if let Some(cluster) = in_cluster {
                     f.write_str(" IN CLUSTER ");
@@ -3863,6 +3868,25 @@ impl<T: AstInfo> AstDisplay for ShowCreateSinkStatement<T> {
 }
 impl_display_t!(ShowCreateSinkStatement);
 
+/// `SHOW [REDACTED] CREATE METRIC SINK <sink>`
+#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
+pub struct ShowCreateMetricSinkStatement<T: AstInfo> {
+    pub metric_sink_name: T::ItemName,
+    pub redacted: bool,
+}
+
+impl<T: AstInfo> AstDisplay for ShowCreateMetricSinkStatement<T> {
+    fn fmt<W: fmt::Write>(&self, f: &mut AstFormatter<W>) {
+        f.write_str("SHOW ");
+        if self.redacted {
+            f.write_str("REDACTED ");
+        }
+        f.write_str("CREATE METRIC SINK ");
+        f.write_node(&self.metric_sink_name);
+    }
+}
+impl_display_t!(ShowCreateMetricSinkStatement);
+
 /// `SHOW [REDACTED] CREATE INDEX <index>`
 #[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
 pub struct ShowCreateIndexStatement<T: AstInfo> {
@@ -5517,6 +5541,7 @@ pub enum ShowStatement<T: AstInfo> {
     ShowCreateSource(ShowCreateSourceStatement<T>),
     ShowCreateTable(ShowCreateTableStatement<T>),
     ShowCreateSink(ShowCreateSinkStatement<T>),
+    ShowCreateMetricSink(ShowCreateMetricSinkStatement<T>),
     ShowCreateIndex(ShowCreateIndexStatement<T>),
     ShowCreateConnection(ShowCreateConnectionStatement<T>),
     ShowCreateCluster(ShowCreateClusterStatement<T>),
@@ -5535,6 +5560,7 @@ impl<T: AstInfo> AstDisplay for ShowStatement<T> {
             ShowStatement::ShowCreateSource(stmt) => f.write_node(stmt),
             ShowStatement::ShowCreateTable(stmt) => f.write_node(stmt),
             ShowStatement::ShowCreateSink(stmt) => f.write_node(stmt),
+            ShowStatement::ShowCreateMetricSink(stmt) => f.write_node(stmt),
             ShowStatement::ShowCreateIndex(stmt) => f.write_node(stmt),
             ShowStatement::ShowCreateConnection(stmt) => f.write_node(stmt),
             ShowStatement::ShowCreateCluster(stmt) => f.write_node(stmt),