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

[mtabebe]

Problem:

The curated metric-sink library runs on every replica and re-renders on
boot, so it cannot be modeled as a durable catalog item the way a user's
CREATE METRIC SINK. That would mean per-boot catalog churn and a
builtin migration every time a definition changes.

Solution:

A new coord/metric_sink.rs installs each CURATED definition on every
replica, mirroring the introspection-subscribe install: bootstrap walks
the existing replicas, replica create installs, replica drop tears down.
Each sink gets a transient GlobalId and is recorded in a
(replica, definition) -> InstalledMetricSink registry once its dataflow
ships, so a definition that fails to plan or optimize leaves no entry behind.
A replica dropped mid-install is caught by the staged validity recheck
before the finish stage ships.

Sequencing is staged, so optimization runs off the coordinator thread.
The finish stage picks the as-of under a read hold and ships the dataflow
targeted at the replica.

The install enforces the definition contract at its own site, because a
curated sink has no plan-time gate: the prefix is validated, a source that
imports a storage collection is rejected (curated SQL must read only
introspection relations, so envd's write frontier never joins the emission
path), and a duplicate registry key restores the first install rather than
shipping a second dataflow.

To build a dataflow from curated SQL, the metric-sink optimizer's input
becomes a MetricSinkFrom::{Id, Query}, mirroring SubscribeFrom.

The flag-off path does not clear existing installs on the next replica
restart: command-history replay re-renders a replica-targeted dataflow
on reconnect, so they persist until the replica is dropped or envd restarts.
The operator's health gauges are labelled with the sink's stable
CuratedMetricSink::name, not its transient GlobalId, so the series stay
identifiable across boots.

Note:

CURATED is empty. Because no definition is installed, the end-to-end pipeline
(install, optimize, pick the as-of under a read hold, ship to the replica, tear
down on replica drop, and the curated label reaching the gauge) does not
run in this PR. The first real definition is next (SQL-556). The tests here
cover only what is reachable without a definition.

Testing:

• An optimizer unit test assembles the Query source path and asserts the
  same shape as the Id path: one MetricSink export over the shaped view,
  the source imported rather than rebuilt.

• A unit test drives plan_source against a debug catalog: it accepts the
  canonical-column contract, and rejects a missing column, a non-trivial
  finishing (checked against the source arity so shaped columns cannot
  desync), and SQL that is not exactly one statement.

• Unit tests cover the sink gauge label: a user sink keeps its GlobalId,
  a curated sink uses its name.

• curated_names_are_unique and curated_definitions_plan guard the static
  list at build time: a duplicate name, which the registry key would collide
  on, or a definition that fails to plan is caught as a failing test rather
  than a boot-time soft-panic. Both iterate an empty list today.

Co-Authored-By: Moritz Hoffmann antiguru@gmail.com

[def-]

QA LLM Review

1. MEDIUM -- Indexed storage inputs bypass the introspection-only guard

src/adapter/src/coord/metric_sink.rs:287

The introspection-only guard treats an empty id_bundle.storage_ids as proof that the curated query has no catalog-backed input, but a storage-backed relation with an available compute index is represented solely as an index import. Such a definition passes the guard and ships, coupling the metric sink to envd-backed frontiers and read capabilities despite this path's stated freshness-isolation contract.

Details

DataflowBuilder::import_into_dataflow prefers every available index and does not add the underlying relation to source_imports in that case (src/adapter/src/optimize/dataflows.rs:326). dataflow_import_id_bundle then puts the chosen index only in compute_ids, leaving storage_ids empty, so the check at lines 292-299 accepts a query over an indexed table or materialized view. The finish stage acquires a hold on that index and the shipped dataflow follows its write frontier, which transitively restores the storage dependency this validation is intended to exclude. Validate catalog provenance before optimization, including the dependency closure of views, or trace each imported index through its on_id and reject it unless it is an introspection-source index over a log collection.

[antiguru]

Seems good. I'll let my agent post a review in a second, but I think this structurally checks out.

[antiguru]

👍 Makes sense, and maybe this should be the default anyway. No need to change the PR, we can easily change this in the future.

[antiguru]

Reviewed the whole diff against the merge-base. The structure is right and mirrors coord::introspection closely enough that the divergences stand out, which is what most of the inline comments are about. Nothing here blocks landing an inert scaffold, but two of the findings are cheaper to fix now than after CURATED is populated, because both are the kind of thing a first definition will hit and neither will surface as a test failure.

The two that matter most. The staged pipeline never extends PlanValidity with the dependencies optimization introduces, which is a step the introspection path takes at the same point and for the reason that applies here too. And the introspection-only contract is enforced by inspecting import kinds after optimization, which decides the question by the target cluster's index layout rather than by what the definition reads, so the same definition passes on mz_catalog_server and soft-panics on every user cluster. Details inline.

A design note worth settling before SQL-556: plan_source plans through the one-shot SELECT path, so finish_maintained never runs on a query that becomes a maintained dataflow. That is the only reason the finishing rejection exists, and the introspection path avoids it by planning as QueryLifetime::Subscribe. Also inline.

On the testing. The PR description is straight about the end-to-end path being unexercised, which I appreciate, but it overstates what the unit tests cover: plan_source_enforces_the_metric_sink_contract has three assertions (valid shape, missing column, ORDER BY) and none for the statement-count rejection the description credits it with, and curated_prefixes_are_valid is not mentioned. The testdrive block is worth calibrating too. With CURATED empty, install_metric_sinks reduces to iterating an empty slice and drop_metric_sinks to an empty range scan, so what it actually asserts is pre-existing user-sink behavior across replica churn. Fine as a harness that starts paying once a definition lands, as long as it is not counted as coverage of this change.

Two things I verified rather than assumed, since both looked like candidate regressions and neither is:

• Dropping the eager import_into_dataflow from the Id path is a genuine cleanup, not a lost import. import_view_into_dataflow walks view.depends_on() and calls import_into_dataflow per leaf, and maybe_reoptimize_imported_views is still called once afterwards. The Id path ends up with the same imports it had before.
• The module doc's claim that a reconnecting replica needs no reinstall holds. Instance::remove_replica produces ERROR_TARGET_REPLICA_FAILED only for subscribes and peeks; sinks come back through rehydrate_replica and command-history replay, and protocol/history.rs advances the replayed as_of under allowed compaction, so a stale as-of cannot reach the replica. Worth keeping that reasoning in the doc, it is not obvious.

The label change is behavior-preserving for user sinks: the default self.sink_id.to_string() and export_sink(self.sink_id, ..) use the same id, so the sink label is unchanged. And ComputeSinkConnection has no protobuf representation, so adding a field is serde-only with no wire migration.

One open question I could not answer from the code. I could not confirm that DROP CLUSTER emits per-replica Dropped implications, and if it does not, drop_metric_sinks never runs on that path and the registry keeps entries under dead ReplicaIds. drop_introspection_subscribes has the identical exposure, so this is pre-existing either way and the leak is bounded memory rather than orphaned collections. Still worth an answer before the list grows, since it decides whether teardown belongs on the replica hook alone.

_{Posted by Claude Code}

[antiguru]

The validity carried into the finish stage never learns about the dependencies optimization introduces. introspection.rs does exactly this at the same point in its pipeline:

let id_bundle = global_mir_plan.id_bundle(cluster_id);let item_ids = id_bundle.iter().map(|id| catalog.resolve_item_id(&id));
validity.extend_dependencies(&catalog, item_ids);

Here validity only holds the resolved_ids from plan_source, which are the items the SQL names. The index imports the optimizer selects are not among them, so an index dropped between this stage and the finish stage passes the recheck and the finish stage ships a dataflow importing a collection that no longer exists.

Not reachable today, since CURATED is empty and the introspection log indexes a definition would import are builtin. It becomes reachable as soon as a definition reads through a droppable index, and the failure is silent up to the point the controller rejects the dataflow.

_{Posted by Claude Code}

[antiguru]

This check tests a property that is not the contract, and the answer it gives depends on which cluster the definition is being installed onto.

dataflow_import_id_bundle splits imports by kind, not by what the query reads (optimize/dataflows.rs:276):

let storage_ids = dataflow.source_imports.keys().copied().collect();let compute_ids = dataflow.index_imports.keys().copied().collect();

import_into_dataflow prefers an index whenever one exists on the cluster and only falls back to import_source when none does. So a definition reading mz_catalog.mz_objects behaves differently per cluster:

• on mz_catalog_server, the builtin index exists, the import is an index import, it lands in compute_ids, and the check passes. envd's write frontier is on the emission path anyway, which is precisely what the check is meant to prevent.
• on any user cluster, no index exists, the import is a source import, and the check soft-panics. Once per cluster.

Same definition, opposite verdict, decided by one cluster's index layout rather than by the definition.

The contract is a statement about what the definition reads, so it wants to be checked against the resolved dependencies' catalog items in plan_source: every dependency must be a CatalogItem::Log or a view over them. That runs once per definition instead of once per replica, fails before any optimization work, and attributes the error to the definition rather than to a cluster.

_{Posted by Claude Code}

[mtabebe]

Ok I have implemented the check, but not in plan_source. The check needs to be done in the coordinator where the CatalogItem is visible.

[antiguru]

mz_sql::plan::plan on a SELECT statement plans with QueryLifetime::OneShot (plan/statement/dml.rs:197), but the result becomes a maintained dataflow. The one substantive consequence is in plan_root_query:

if lifetime.is_maintained(){
expr.finish_maintained(&mut finishing, group_size_hints);}

Under OneShot that never runs, so the finishing is left beside the expression instead of folded into it, which is the only reason the rejection below has to exist at all. A maintained lifetime would fold it in and the question would not arise. allow_show also differs between the two lifetimes, so a curated definition containing SHOW ... plans here and would not under a maintained lifetime.

The introspection path plans its SQL as QueryLifetime::Subscribe, which is the right shape for a continuously-maintained collection. Planning through plan_root_query with a maintained lifetime rather than through Plan::Select would match it and drop the finishing check.

Separately, the error message is inaccurate. is_trivial also fails on a non-identity projection (expr/src/relation.rs:3544):

self.limit.is_none() && self.order_by.is_empty() && self.offset == 0
&& self.project.iter().copied().eq(0..arity)

so a definition with no ORDER BY, LIMIT, or OFFSET anywhere can still land on this branch and be told to remove clauses it does not have. Worth naming the projection case in the message, since a definition with a FROM clause is the shape every real definition will take and the tests only cover SELECT-without-FROM.

_{Posted by Claude Code}

[mtabebe]

Done, I changed the planning to be a plan with the source as SUBSCRIBE, so it goes through the right life cycle.

[antiguru]

install_metric_sink calls plan_source inside this loop, so a definition is parsed, resolved, described, and planned once per replica. The plan depends only on the catalog, never on the replica: plan_source takes no replica argument, and the replica enters the picture only at PlanValidity and the optimizer.

At bootstrap that is |CURATED| x |replicas| SQL plans run serially on the coordinator thread, before the message loop starts draining the staged installs, and the same multiplier applies again on every replica create. Planning once per definition and cloning the resulting HirRelationExpr per replica removes the multiplier and keeps a definition's plan errors from being reported once per replica.

_{Posted by Claude Code}

[antiguru]

Two things about the collision path.

The comment names a cause that curated_names_are_unique already rules out. Two distinct definitions cannot collide on (replica_id, name), since the test fails the build on a duplicate name. The reachable cause is install_metric_sinks running twice for one replica, which is a different bug with a different fix, so the comment sends the next reader looking in the wrong place.

The guard also fires as late as it can. Nothing consults self.metric_sinks at the install site, so a double install pays a full plan_source plus a full optimize per definition and only then discovers the key is taken. A registry check in install_metric_sink before plan_source makes the duplicate cheap and puts the diagnostic next to the caller that caused it. Keeping this one as the backstop is still right.

_{Posted by Claude Code}

[antiguru]

Sharing this between the two features is right, and the doc comment on it is accurate. Worth spelling out one implication in the metric-sink module doc, though: the set is unfiltered, so it includes every replica of every user cluster alongside the system ones.

That means each curated definition becomes a dataflow, with its arrangements, on customer compute, charged to the customer's cluster. For introspection subscribes that cost is already accepted. For curated metric sinks it is a new charge that neither the module doc nor the PR description mentions, and it scales with |CURATED|. Better to state the intent now than to discover it when the list grows.

_{Posted by Claude Code}

[antiguru]

This comment contradicts the session state the file already set up. Line 31 does SET cluster_replica = r1 and nothing resets it, so the session has picked a replica for every read in the file, including these. UntargetedLogRead cannot fire here.

Which means the interesting window is assertable: with two replicas up, cluster_replica = r1 reads r1's registry and the count should still be 1. Asserting there, and again after the scale-down, would give the block a claim about replica churn that it currently does not make. As written it reads the same registry twice with the same expectation and the churn between them is unobserved.

_{Posted by Claude Code}

[antiguru]

Most of the suite takes the replica size from ${arg.default-replica-size} with a set-arg-default at the top of the file, which is what lets the multi-size testdrive configurations vary it. Hardcoding it here pins this cluster to one size regardless of how the run is configured.

_{Posted by Claude Code}

[mtabebe]

Regarding drop, I think it is safe.

The way I read the code is: remove_clusters deletes the cluster's replicas, so a DROP CLUSTER diff carries a per-replica ClusterReplica::Dropped. This then drives drop_replica → drop_metric_sinks for each. Nothing is left under dead ReplicaIds. I think it is the same pattern as drop_introspection_subscribes.

[def-]

QA LLM Review

1. MEDIUM -- ensure_reads_only_logs does not see relations read through SQL-implemented functions

src/adapter/src/coord/metric_sink.rs:477

The new introspection-only gate walks only the ids mz_sql::names::resolve found in the statement's AST, and treats a CatalogItem::Func as a leaf with no data dependency. A curated definition calling any of the SQL-implemented builtins whose body reads a catalog relation therefore passes the gate while its dataflow imports that relation, which is the envd-frontier coupling the gate exists to prevent.

Details

plan_source already has the missing ids in hand and discards them: mz_sql::plan::plan returns them as its second value (metric_sink.rs:512), populated by sql_impl at src/sql/src/func.rs:369 when it resolves a function body. mz_sql keeps them out of resolved_ids deliberately, on the grounds that they are "implementation details of the functions, not real dependencies of the statement" (src/sql/src/plan/statement.rs:476). That holds for a one-shot statement; it does not hold here, because the body is inlined into the HIR and its Gets become real dataflow imports.

Concretely, pg_get_viewdef expands to SELECT definition FROM mz_catalog.mz_views ... and pg_table_is_visible to a join over mz_catalog.mz_objects and mz_catalog.mz_schemas (src/sql/src/func.rs:2712, :2822). A definition using one of these gets, per cluster: on mz_catalog_server the builtin index makes it an index import, so metric_sink_finish's storage_ids backstop passes too and the sink ships coupled to envd; on a user cluster it becomes a source import, so the backstop fires after a full plan and optimize, once per replica, and soft-panics (a hard panic wherever soft assertions are on). PlanValidity misses the same relations, though that half is inert while they are builtins.

Fix: bind the second return value of mz_sql::plan::plan and union it into dependencies before ensure_reads_only_logs runs, so the closure covers what the dataflow actually reads rather than what the SQL names. The general weakness in CatalogEntry::uses for functions, which the CatalogItem::Func arm inherits, is already tracked in MaterializeInc/database-issues#9936; the discarded ids at this call site are not.

Latent until the first CURATED entry lands.