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

src/adapter/src/coord.rs

@@ -200,6 +200,7 @@ use crate::coord::appends::{
 use crate::coord::caught_up::CaughtUpCheckContext;
 use crate::coord::id_bundle::CollectionIdBundle;
 use crate::coord::introspection::IntrospectionSubscribe;
+use crate::coord::metric_sink::{CuratedMetricSink, InstalledMetricSink, PlannedMetricSink};
 use crate::coord::peek::PendingPeek;
 use crate::coord::statement_logging::StatementLogging;
 use crate::coord::timeline::{TimelineContext, TimelineState};
@@ -243,6 +244,7 @@ mod indexes;
 mod info_metrics;
 mod introspection;
 mod message_handler;
+mod metric_sink;
 mod privatelink_status;
 mod sql;
 mod validity;
@@ -449,6 +451,10 @@ pub enum Message {
         span: Span,
         stage: IntrospectionSubscribeStage,
     },
+    MetricSinkStageReady {
+        span: Span,
+        stage: MetricSinkStage,
+    },
     SecretStageReady {
         ctx: ExecuteContext,
         span: Span,
@@ -575,6 +581,7 @@ impl Message {
             Message::IntrospectionSubscribeStageReady { .. } => {
                 "introspection_subscribe_stage_ready"
             }
+            Message::MetricSinkStageReady { .. } => "metric_sink_stage_ready",
             Message::SecretStageReady { .. } => "secret_stage_ready",
             Message::ClusterStageReady { .. } => "cluster_stage_ready",
             Message::DrainStatementLog => "drain_statement_log",
@@ -1181,6 +1188,36 @@ pub struct IntrospectionSubscribeFinish {
     replica_id: ReplicaId,
 }
 
+#[derive(Debug)]
+pub enum MetricSinkStage {
+    Optimize(MetricSinkOptimize),
+    Finish(MetricSinkFinish),
+}
+
+#[derive(Debug)]
+pub struct MetricSinkOptimize {
+    validity: PlanValidity,
+    definition: &'static CuratedMetricSink,
+    /// The transient id of the sink's compute export. Recorded in
+    /// [`Coordinator::metric_sinks`] once the finish stage ships the dataflow.
+    sink_id: GlobalId,
+    /// The planned `source_sql`, and the shape it produces.
+    expr: HirRelationExpr,
+    desc: RelationDesc,
+    cluster_id: ComputeInstanceId,
+    replica_id: ReplicaId,
+}
+
+#[derive(Debug)]
+pub struct MetricSinkFinish {
+    validity: PlanValidity,
+    definition: &'static CuratedMetricSink,
+    sink_id: GlobalId,
+    global_lir_plan: optimize::metric_sink::GlobalLirPlan,
+    cluster_id: ComputeInstanceId,
+    replica_id: ReplicaId,
+}
+
 #[derive(Debug)]
 pub enum SecretStage {
     CreateEnsure(CreateSecretEnsure),
@@ -2117,6 +2154,14 @@ pub struct Coordinator {
     hydration_history_replica_cursor: Option<ReplicaId>,
     /// Hydration-history sweep owned by the coordinator while one is in flight.
     hydration_history_sweep: Option<AbortOnDropHandle<()>>,
+    /// The curated metric sinks installed on each replica.
+    ///
+    /// Keyed replica-first so a replica's installs form one contiguous range: teardown on replica
+    /// drop is the only lookup that is not by exact key.
+    metric_sinks: BTreeMap<(ReplicaId, &'static str), InstalledMetricSink>,
+    /// Curated metric-sink plans, cached per definition so each is planned once rather than once
+    /// per replica. See [`Coordinator::plan_metric_sink`].
+    metric_sink_plans: BTreeMap<&'static str, PlannedMetricSink>,
 
     /// Locks that grant access to a specific object, populated lazily as objects are written to.
     write_locks: BTreeMap<CatalogItemId, Arc<tokio::sync::Mutex<()>>>,
@@ -3112,6 +3157,9 @@ impl Coordinator {
         // Initialize unified introspection.
         self.bootstrap_introspection_subscribes().await;
 
+        // Install the curated metric sinks on every replica.
+        self.bootstrap_metric_sinks().await;
+
         info!(
             "startup: coordinator init: bootstrap: migrate builtin tables in read-only mode complete in {:?}",
             final_steps_start.elapsed()
@@ -3909,9 +3957,12 @@ impl Coordinator {
 
                                 // MIR ⇒ MIR optimization (global)
                                 let metric_sink_plan = optimize::metric_sink::MetricSink::new(
-                                    entry.name().clone(),
-                                    metric_sink.from,
+                                    self.catalog()
+                                        .resolve_full_name(entry.name(), None)
+                                        .to_string(),
+                                    optimize::metric_sink::MetricSinkFrom::Id(metric_sink.from),
                                     metric_sink.prefix.clone(),
+                                    None,
                                 );
                                 let global_mir_plan = optimizer.optimize(metric_sink_plan)?;
                                 let optimized_plan = global_mir_plan.df_desc().clone();
@@ -5341,6 +5392,8 @@ pub fn serve(
                     introspection_subscribes: BTreeMap::new(),
                     hydration_history_replica_cursor: None,
                     hydration_history_sweep: None,
+                    metric_sinks: BTreeMap::new(),
+                    metric_sink_plans: BTreeMap::new(),
                     write_locks: BTreeMap::new(),
                     deferred_write_ops: BTreeMap::new(),
                     pending_writes: Vec::new(),

src/adapter/src/coord/catalog_implications.rs

@@ -1688,6 +1688,7 @@ impl Coordinator {
 
         self.install_introspection_subscribes(cluster_id, replica_id)
             .await;
+        self.install_metric_sinks(cluster_id, replica_id).await;
     }
 }
 

src/adapter/src/coord/ddl.rs

@@ -719,6 +719,7 @@ impl Coordinator {
 
     pub(crate) fn drop_replica(&mut self, cluster_id: ClusterId, replica_id: ReplicaId) {
         self.drop_introspection_subscribes(replica_id);
+        self.drop_metric_sinks(replica_id);
 
         self.controller
             .drop_replica(cluster_id, replica_id)

src/adapter/src/coord/introspection.rs

@@ -98,18 +98,27 @@ impl IntrospectionSubscribe {
 }
 
 impl Coordinator {
+    /// Every `(cluster, replica)` pair currently in the catalog.
+    ///
+    /// The set a per-replica feature (introspection subscribes, curated metric sinks) must install
+    /// onto the replicas that already exist when the coordinator starts. Shared so those callers
+    /// cannot drift on what "all replicas" means.
+    pub(super) fn all_cluster_replicas(&self) -> Vec<(ClusterId, ReplicaId)> {
+        self.catalog
+            .clusters()
+            .flat_map(|cluster| {
+                cluster
+                    .replicas()
+                    .map(move |replica| (cluster.id, replica.replica_id))
+            })
+            .collect()
+    }
+
     /// Installs introspection subscribes on all existing replicas.
     ///
     /// Meant to be invoked during coordinator bootstrapping.
     pub(super) async fn bootstrap_introspection_subscribes(&mut self) {
-        let mut cluster_replicas = Vec::new();
-        for cluster in self.catalog.clusters() {
-            for replica in cluster.replicas() {
-                cluster_replicas.push((cluster.id, replica.replica_id));
-            }
-        }
-
-        for (cluster_id, replica_id) in cluster_replicas {
+        for (cluster_id, replica_id) in self.all_cluster_replicas() {
             self.install_introspection_subscribes(cluster_id, replica_id)
                 .await;
         }

src/adapter/src/coord/message_handler.rs

@@ -231,6 +231,9 @@ impl Coordinator {
             Message::IntrospectionSubscribeStageReady { span, stage } => {
                 self.sequence_staged((), span, stage).boxed_local().await;
             }
+            Message::MetricSinkStageReady { span, stage } => {
+                self.sequence_staged((), span, stage).boxed_local().await;
+            }
             Message::ExplainTimestampStageReady { ctx, span, stage } => {
                 self.sequence_staged(ctx, span, stage).boxed_local().await;
             }