compute: add a lifecycle event log for compute exports - #38403
Conversation
Design: a lifecycle event log instead of wide timestamp columnsWriting up a design we converged on offline, which supersedes the approach in this PR. The code here is sound, but the shape it extends is not the one we want, and the reasons are worth recording before the stacked The problem with wide columnsTwo structural issues surface as soon as the write stages are added. The first is that the stages do not share a grain. The second is that a timestamp column cannot say why the next stage has not happened. ShapeOne append-only relation, per replica, in memory.
Indexes emit the first three and stop, which is the index degeneracy of the lifecycle falling out of the model rather than being special cased. Subscribes and Why
|
67183be to
93ee52b
Compare
antiguru
changed the title
93ee52b to
c906ea5
Compare
QA LLM Review1. HIGH -- Adding a builtin log also moves
|
c906ea5 to
aff3e16
Compare
|
Confirmed and fixed in I had found the Also took the test suggestion, and it is sharper than it first looks. For the record on the rest of this PR's CI: the five Generated by Claude Code |
QA LLM Review1. MEDIUM -- The back-filled
|
aff3e16 to
581179d
Compare
|
Confirmed and fixed in Split it into The part worth dwelling on: I wrote the comment "The lifecycle log needs the same back-fill" directly above a call that did not perform the same back-fill. The Two notes on the surrounding state, so the next review has the current picture: The I hand-edited eight goldens for this relation and have since found two errors in my own edits, so I am not going to keep hand-editing them. A local Generated by Claude Code |
d183b3b to
1a6f46f
Compare
The compute change took a different shape than the one recorded here. Rather than redefining `hydrated_at` and adding a `written_at` column, #38403 adds an append-only lifecycle log and leaves `mz_compute_hydration_times_per_worker` untouched. So there is no semantic shift for rows this collector has already written, and no ordering constraint between that change and enabling collection in production. What survives is the guidance that recording gates on hydration rather than on a write stage, and a note that the event log carries the as-of, which this table does not, and without which a duration does not say how much work was done. Ref: SQL-644
Compute is adding an append-only lifecycle log for the same stages (#38403) rather than more timestamp columns, and it leaves `mz_compute_hydration_times_per_worker` untouched. So nothing this collector has recorded changes meaning, and there is no ordering constraint between that work and enabling collection here. Notes what moving onto that log would buy, and the one piece of guidance that outlives the current shape: recording gates on hydration, not on a write stage, because a replacement runs read-only until cutover and would otherwise never be recorded at all. Also notes that the event log carries the dataflow's as-of, which this table does not, and without which a duration does not say how much work was done. Ref: SQL-644
Compute is adding an append-only lifecycle log for the same stages (#38403) rather than more timestamp columns, and it leaves `mz_compute_hydration_times_per_worker` untouched. So nothing this collector has recorded changes meaning, and there is no ordering constraint between that work and enabling collection here. Notes what moving onto that log would buy, and the one piece of guidance that outlives the current shape: recording gates on hydration, not on a write stage, because a replacement runs read-only until cutover and would otherwise never be recorded at all. Also notes that the event log carries the dataflow's as-of, which this table does not, and without which a duration does not say how much work was done. Ref: SQL-644
QA LLM Review1. MEDIUM -- Empty completion suppresses hydration for successfully completed exports
The unconditional return for an empty progress frontier treats successful completion the same as cancellation. Input-free exports such as an index on DetailsThe constant renderer builds an inputless Emptiness does not encode why progress ended. A |
QA LLM Review1. MEDIUM -- A shard-wide upper cannot attribute
|
Compute is adding an append-only lifecycle log for the same stages (#38403) rather than more timestamp columns, and it leaves `mz_compute_hydration_times_per_worker` untouched. So nothing this collector has recorded changes meaning, and there is no ordering constraint between that work and enabling collection here. Notes what moving onto that log would buy, and the one piece of guidance that outlives the current shape: recording gates on hydration, not on a write stage, because a replacement runs read-only until cutover and would otherwise never be recorded at all. Also notes that the event log carries the dataflow's as-of, which this table does not, and without which a duration does not say how much work was done. Ref: SQL-644
QA LLM Review1. MEDIUM -- The new dataflow link is not a many-to-one foreign key
Details
|
Record each compute export's lifecycle as an append-only log,
`mz_introspection.mz_compute_lifecycle_events_per_worker`, rather than as more
timestamp columns on the hydration time relation.
Two things stop timestamp columns from carrying the lifecycle. The stages do not
share a grain: `installed`, `started` and `hydrated` are per-worker facts, since
each worker hydrates its own fragment of the dataflow, while whether the output
is durable is a property of the sink as a whole, maintained on one elected
worker. And a timestamp cannot say why the next stage has not happened, so a
NULL cannot tell a replacement materialized view waiting for a cutover apart
from an index that will never write.
export_id text not null
worker_id uint8 not null
event text not null
occurred_at timestamptz not null
reason text nullable
details jsonb nullable
`installed`, `started` and `hydrated` are logged by every worker. The write
stages are logged only by the worker that maintains the sink's shared write
frontier, so they appear once per object and the row records which worker was
elected. An index emits the first three and stops, which is the index degeneracy
of the lifecycle falling out of the model rather than being special-cased.
`hydrated` reads the dataflow's own progress frontier, the compute probe, not
the reported output frontier. The output frontier folds in the write frontier,
which makes it a measure of durability, and for a sink-backed collection it is
not even uniform across workers: `mint` clears the shared frontier on every
non-elected worker, where it is the empty antichain and contributes nothing to
the meet. `mz_compute_hydration_times_per_worker` is unchanged, so `hydrated_at`
and `time_ns` keep reporting exactly what they reported before, and the new
relation carries the dataflow reading alongside.
The write stages are gated on hydration. Before it the sink has produced nothing
and read-only mode is holding nothing back, and every collection starts
read-only, so reporting a block from installation would put a `write_blocked`
and a `write_unblocked` on essentially every materialized view, both ahead of
`hydrated`. Gating also keeps `written` ordered after `hydrated`, which it is not
otherwise: `apply_refresh` rounds a `REFRESH` materialized view's frontier up to
the next refresh time off its input frontier, before the dataflow computes
anything, so the sink writes an empty batch for the pre-refresh window and the
shard's upper passes the as-of while the dataflow is still hydrating. That also
means a refresh schedule advances writing rather than blocking it, so there is no
`refresh` cause for `write_blocked` to report.
`details` carries the dataflow's as-of, which every stage is defined relative to:
without it the interval between two stages says nothing about how much work was
done, since a replacement materialized view with a far behind as-of is a
completely different amount of work at the same duration.
Part of CPU-226
Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
`observe_hydration` took a bool, leaving the choice of frontier and the handling of an empty one to its two call sites. Those call sites had already diverged. `report_frontiers` compares the as-of against a probe or write frontier with no emptiness check; `process_subscribes` compares it against a batch upper and guards with `matches!(response, SubscribeResponse::Batch(_))`, whose comment claims that filtering out `DroppedAt` is enough to stop a cancelled subscribe from reading as hydrated. It is not enough. A subscribe signals completion by sending a batch at the empty frontier, so the completion batch passes the `matches!` and the empty antichain is the maximum of the order. `SUBSCRIBE ... UP TO x AS OF x` is legal, only attaching an `EqualSubscribeBounds` notice, and it emits no rows at all: its `up_to` filter admits no times. It then logs `hydrated` for a dataflow that computed nothing. Take the frontier itself and decide inside, so there is one definition of what counts as progress. An empty frontier reports completion, not hydration, whether it arrives as `DroppedAt`, as a completion batch, or from anywhere else. A subscribe that genuinely hydrated has already logged it from the preceding batch, so nothing is lost. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
`written` compared the as-of against the sink's write frontier. That frontier is
the output shard's upper, a property of the shard rather than of this replica,
and `as_of_selection::apply_downstream_storage_constraints` bounds the as-of to
one step below the upper for a non-empty storage export. So for any shard that
already holds data the comparison is true from the moment the dataflow is
installed, and the stage says only that somebody once wrote the output.
The read-only gate did not fix this, it deferred it by one poll. A replica that
is permitted to write is not thereby the replica that wrote. Three cases got a
`written` for appending nothing:
* a replica added by raising the replication factor,
* a replica whose process restarted,
* a read-only replica at cutover, where `write_unblocked` and `written` land in
the same call and the interval between them is always zero.
The third is the case the relation exists to measure, and the previous NOTE at
this site claimed the guard prevented exactly what it permitted.
Latch the frontier the first time writes are permitted and require it to advance
past that baseline. The latch is deliberately before the hydration gate rather
than after it: latching later would fold this replica's own early writes into the
baseline and never report them, while latching before the block is lifted would
measure against an upper the previous writer goes on to advance.
`written` now means that this replica's sink advanced the shard beyond where it
stood when the replica was allowed to write. For a fresh materialized view, whose
as-of is not stepped back, that is the first real append, unchanged from before.
Part of CPU-226
Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
The relation promises at most one row per export, worker and stage, and a reader relies on it: an event's `occurred_at` is taken directly, without aggregating first. That promise was kept by three unrelated mechanisms, none of them in the demux that owns the log. `CollectionState::logged_stages` covered four stages, `hydration_timestamps.started_at` covered `started`, and `installed` relied on `handle_export` running once per export. `log_lifecycle_at` pushed unconditionally, and `CollectionLogging::log_lifecycle` is public and accepts any stage, including the two the demux already emits from its own events. A caller passing either produced a duplicate row that nothing rejected. The doc comment stated the contract, the type did not. Key the retained rows by stage and ignore a stage already logged. The bound on the map is now structural rather than incidental, the back-fill of `started` in `handle_hydration` no longer needs to know whether `handle_hydration_start` already ran, and a caller that cannot cheaply tell whether it has reported a stage does not have to. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
Three cleanups to the boundary the write lifecycle stages depend on. `frontier_owner` lived in the v1 materialized view sink and was named for a property v1 does not own. The election belongs to the persist sink protocol that both implementations share, so it moves to `crate::sink`, and all four call sites now spell it the same way. `materialized_view_v2` had been reaching for it twice in one file under two different paths. `metric_sink` carries a byte-identical expression and must keep it. Its shared frontier is written by every worker, before the early return for the inactive ones, so it has no elected owner at all, and the worker it does elect is the one that folds metrics into the registry. The obvious next cleanup would be to call `frontier_owner` there and silently couple two independent elections, so both sites now say why they are separate. `sink_write_frontier` and `owns_sink_frontier` were two public fields whose coupling lived only in a doc comment. Three sinks set the frontier and two set the flag. A future sink that sets the frontier and forgets the flag loses every write stage with no error, and one that sets it wrongly reports having written everything immediately, because a non-owning worker's copy is cleared to the empty antichain. One setter takes both, and the fields are now private. Also correct the `ExportState::as_of` comment: a dataflow retained across reconciliation logs no new `Export` event, so the field holds the as-of it was installed with rather than the collection's current one. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
A dataflow's exports share one suspension token, so its computation begins only once every export has been scheduled. `handle_schedule` says so in a comment and then reports hydration start for the single export the command named. For a dataflow exporting A and B, scheduled at t1 and t2, the dataflow starts at t2 but A's `started` reads t1. A claims to have started while its dataflow was still suspended, and `hydrated - started` overstates the compute time by t2 - t1. Report the start only on the token release that actually unsuspends the dataflow, and report it for every export of that dataflow. The start is a property of the dataflow, so all its exports share the instant, and now they share it because that is when computation began rather than by approximation. This is latent today: nothing in production ships more than one export per dataflow, since `export_index` and `export_sink` each insert one. It affects `mz_compute_hydration_times_per_worker.started_at` equally. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
Rejecting the empty frontier in `observe_hydration` was wrong. An inputless collection, such as an index or materialized view on `SELECT 1`, emits its rows and drops its capability, so its frontier goes from the initial time straight to the empty antichain without ever holding a non-empty value beyond its as-of. Rejecting empty leaves such an export permanently unhydrated in the lifecycle relation, and because the write stages are gated on `hydrated`, a materialized view of that shape reports no write stage ever. It also contradicted the durability reading. `CollectionState::hydrated` compares the as-of against the reported output frontier with no emptiness check, so the existing relation calls these exports hydrated. The two relations would have disagreed for the same object. Emptiness therefore cannot distinguish completion from cancellation, and a caller that can observe a dataflow ending without computing its as-of has to exclude that itself. `process_subscribes` is the one such caller: `DroppedAt` carries the empty antichain for a subscribe cancelled mid-hydration, so it keeps the batch check that excludes it. A subscribe that runs to its `up_to` also finishes with an empty upper, but it carries that in a batch and did compute through its as-of. The frontier-taking signature stays. Passing the frontier rather than a bool is what gives the emptiness question a single answer instead of one per call site. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
The latched baseline did not work, for two independent reasons. It latched a placeholder. `mint` initializes the shared sink frontier to `Antichain::from_elem(Timestamp::MIN)` and only fills in the real upper once its `persist_frontiers` stream has read it back. The latch runs from the first `report_frontiers` poll that sees writes permitted, which is normally before that, so the baseline was `[MIN]` and any real upper compared greater. `written` fired at once, which is what the baseline was meant to prevent. And no reading of that frontier can attribute a write. Every replica's `mint` reads the shard's upper back from persist into its own shared frontier, so the frontier advances on every replica when any one of them wins the append. A baseline latched from it, placeholder or not, says only that the shard moved while this replica was eligible to write. Go back to comparing against the as-of and state the semantics instead of implying stronger ones. `written` means the output is durable through the as-of and this replica was permitted to write. Replicas of a cluster produce identical batches and race to append, so which replica won is not an operationally meaningful question, but it does mean `written` lands with `hydrated` on a restarted or scaled-out replica and with `write_unblocked` at a cutover, and a reader has to know that. Real attribution needs a signal from this replica's own append path. That has to cross workers, because `next_append_worker` rotates independently of `sink::frontier_owner`, so it is left as a TODO rather than approximated. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
…nator A consumer rolling the per-worker relation up per object has to know whether every worker has reached a stage. `mz_compute_hydration_times_per_worker` lets it ask without knowing the worker count, because it holds one row per export and worker from installation with a nullable `time_ns`, so `count(*) = count(time_ns)` is the test. The introspection subscribe for `ComputeHydrationTimes` uses exactly that. An append-only log has no NULLs to count: a worker that has not hydrated has no `hydrated` row. The equivalent is `installed`, which every worker logs unconditionally when the export is created, so the count of `installed` events is the number of workers reporting on that export. That makes it a contract the relation owes its readers rather than an incidental property of the implementation, and it was written down nowhere. Say it in the ontology description, where a reader of the relation will find it, and in the design doc alongside the grain rules, together with the point that which stages have which grain is fixed by the vocabulary rather than varying per object or cluster, so a consumer never needs a catalog join to interpret a count. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
The design doc argued that withholding `written` while writes are blocked keeps another writer's progress from being attributed to this replica. That is true of a blocked replica and says nothing about a permitted one, which leaves the reader with a stronger impression of the stage than it earns. Every replica's `mint` reads the output shard's upper back from persist into its own shared frontier, so the frontier advances on every replica when any one of them wins the append. State that, state the two cases where the implied interval is therefore zero, `written` landing with `hydrated` on a restarted or scaled-out replica and with `write_unblocked` at a cutover, and record why the obvious tightening does not work: a frontier latched when writes are first permitted sees the same concurrent advance, and the value available at the first observation is usually the placeholder `mint` starts from rather than a real upper. Attribution needs a signal from the replica's own append path, which has to cross workers because `next_append_worker` rotates independently of the frontier owner. That goes under "Follow-up work" with the open question of whether the distinction earns the machinery, given that the racing replicas append identical batches. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
The stages do not all describe the same object. `hydrated` and the write stages belong to one export: hydration reads that export's own progress, and the write stages read its sink. `installed` and `started` describe the dataflow, which can maintain more than one export, and `started` is the instant that dataflow was unsuspended, shared by every export it maintains. Keying the relation by export is still right. The identifier a consumer has is the export id, which is what `mz_objects.id` holds, while dataflow ids are per worker and internal to a replica. A dataflow-grain relation could not hold hydration or the write stages at all, so splitting along the seam would produce two relations instead of one and make the most basic question a join through the export-to-dataflow map. Carry the dataflow id as a column instead. The events whose cause is dataflow-wide become recognizable as such, `SELECT DISTINCT dataflow_id, event, occurred_at` recovers the dataflow-level facts, and the redundancy is visible rather than implied. The relation also ends up carrying the export-to-dataflow mapping that `mz_compute_exports_per_worker` holds, for eight bytes a row. No new migration step. `make_mz_indexes` inlines each log's `index_by` column names, so `mz_indexes` moves again, but `MigrationStep::replacement` records only a version and an object, and the step this change already added at the current dev version covers any further change to that SQL within the same version. `make_mz_sources` inlines no columns, so it does not move. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
`lifecycle_event_in_dataflow` declared `dataflow_id` a many-to-one foreign key into `dataflow_global_id_per_worker.id`, which it is not. `ComputeLog::DataflowGlobal` keys that relation on `(id, worker_id, global_id)`, one row per object rendered in the dataflow, so the advertised join both crosses workers and multiplies every lifecycle event by the number of rendered objects. Adding `worker_id` through `extra_key_columns` would not fix it, because the fan-out over `global_id` remains: the relation is a dataflow-to-object mapping, not a dataflow entity, and no ontology entity is keyed by `(dataflow_id, worker_id)` for the link to point at. This metadata is acted on rather than merely read. `mz_ontology_link_types` is what agents are pointed at to discover join paths, so a wrong cardinality produces a query that silently fans out instead of a description that merely reads oddly. Remove the link and put the join guidance where it cannot mislead a planner: the column description now says dataflow ids are worker-scoped, names `(dataflow_id, worker_id)` against `mz_compute_exports_per_worker` as the way to reach a dataflow's exports, and warns that the global-ids relation fans out. The surviving `lifecycle_event_of` link is `MapsTo`, which carries no cardinality and so claims nothing false, but it joins two per-worker relations on `export_id` alone. Its `note` now records that the exact join is `(export_id, worker_id)`. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
The description claimed dataflow ids are worker-scoped and only meaningful paired with `worker_id`. They are not. Each worker assigns indices from a counter advanced by the same command sequence, so a dataflow carries the same index on every worker of a replica, and `dataflow_id` identifies the dataflow rather than a per-worker artifact. It is still a replica-local index and not a catalog id, which is the part that matters for a consumer. Dropping the `lifecycle_event_in_dataflow` link remains right, for the other reason given: `mz_compute_dataflow_global_ids_per_worker` holds one row per object rendered in a dataflow, so `dataflow_id` is not unique there even within a single worker, and a foreign key claiming otherwise makes a generated join fan out. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
`cluster.slt` lists each per-replica introspection index's key columns with their positions, so a column added to an unkeyed log shifts every position after it. `cockroach/srfs.slt` generates a series over each relation's column count, so it gains a row per instance. Both were missed when the column was added. The position listing in `cluster.slt` is a third section of that file, distinct from the two counts already updated there, and searching the file for the relation's name finds it only if the search is not anchored on those counts. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
Four corrections, none changing what the relation reports. `sink_write_frontier` was left public. Its own doc comment says it is set through `set_sink_write_frontier` together with `owns_sink_frontier`, and the point of that setter is that the pairing cannot be forgotten, which a public field defeats. `report_frontiers` is the only other reader and is in the same module. `handle_schedule` scanned every collection to find the dataflow's exports. A `Schedule` arrives once per dataflow, so a replica starting N dataflows made N full traversals of the collection map on the timely worker thread, quadratic in exactly the phase where start-up latency is the thing being measured. Two strong references to the dataflow index mean the named collection is the only export, so take that case directly. `drop_collection` releases a suspension token without the check `handle_schedule` now makes, so for a multi-export dataflow it can be the release that unsuspends the computation while reporting no start. Unreachable while every dataflow reaching a replica has exactly one export, which `SequentialHydration::absorb_command` requires, but silent divergence between two paths that release the same token deserves a note rather than nothing. `CollectionLogging::log_lifecycle` still told callers the demux does not deduplicate. It does, since the retained rows became keyed by stage, and two handlers rely on it. A caller trusting the stale comment would add a redundant guard, or trust it in the other direction. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
Downstream work is being built on these relations now, so what is frozen and what may still move has to be explicit rather than inferred from the current implementation. Inferring it is how a consumer ends up depending on a closed `event` vocabulary that a later stage extends, or on `hydrated_at` meaning what the lifecycle log's `hydrated` means. Frozen: the six event values and their meanings, which events have which grain, `installed` as the all-workers denominator, `occurred_at` as a wallclock instant carrying its worker's anchor, `hydrated` as always the dataflow-progress reading and `written` as always durable-and-permitted, `hydrated_at` and `time_ns` staying the demux-computed durability reading rather than becoming a view, the `(export_id, worker_id)` join between the two relations, and a lifecycle row implying a live export. Open: new `event` values, new `reason` values, new `details` keys. The invariant tests assert closed vocabularies for what exists today, which is not the same as promising the vocabularies are closed forever. The corollary for `written` gets its own paragraph, since it is the one term a scoped follow-up could plausibly redefine. Attributing a write to the replica that performed it arrives as a new stage rather than a redefinition, so a consumer reading `written` today reads the same thing afterwards. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
The comment justified retaining the emitted rows by saying re-deriving them at drop time would risk drifting from what was inserted. That overstates it: `pack_lifecycle_update` is a pure function of the export id, the worker id, the dataflow index, the stage, the instant, and the as-of, all of which outlive the insert, so re-deriving would reproduce the rows today. The real justification is that retaining them makes the retraction correct by construction rather than contingent on the packer staying pure, and that the cost of the guarantee is affordable. Say both, with the number: the relation declares no key, so each entry holds a whole row, around a kilobyte per export per worker once every stage is reached. Also name the alternative and its trade, so a reader who hits a memory problem does not have to rediscover it. Keeping each stage's `occurred_at` and repacking at drop time costs a small fraction of the footprint and gives up the guarantee. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
The compatibility contract promised consumers they would never see a lifecycle row whose `export_id` had left `mz_objects`. Nothing provides that. `DROP` returns once the catalog transaction commits, and `drop_collections` removes controller state and dispatches to the instance without waiting for the replica. The retraction then has to arrive at the replica as an empty `AllowCompaction`, be logged by the demux, and travel through the introspection subscribe and a separate storage append. There is no ordering that makes the catalog row outlive the lifecycle rows. A consumer that took the promise at face value and filtered with an inner join against the catalog would drop the tail of every episode, which is the part worth having. State the asynchrony instead. It is not specific to this relation: `mz_compute_hydration_times_per_worker` rows also disappear only when the replica retracts them, so a consumer already had to tolerate this. The promise was the new thing, and it was wrong. The testdrive orphan check keeps working, since it sits in the retried section and converges, but its comment read as though the retraction were synchronous, which is the misreading that produced the bad contract line. Say that the retry is doing real work there. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
`hydrated` was the one word the two readings still shared, and sharing it kept the ambiguity the lifecycle log exists to remove. `hydrated_at` in `mz_compute_hydration_times_per_worker`, `CollectionState::hydrated` and `mz_hydration_statuses` are all the durability reading, taken from a frontier that folds in the write frontier, so for an object that sinks to persist they mean something the log's stage does not. A consumer reading `hydrated` in one place and `hydrated` in the other had no way to know they differed. Name the stage for what actually happened. `snapshot_complete` is the dataflow's own progress passing its as-of, and it means the same thing for an index and for a materialized view. The durability term in the log stays `written`. After this the two readings share no vocabulary at all, which is the property that makes them hard to confuse. `observe_hydration` becomes `observe_snapshot` for the same reason: it sat immediately beside `hydrated` and `set_hydrated`, which report the other reading. Mechanical elsewhere: the stage enum and its `event` string, the ontology description, the testdrive vocabulary and assertions, and the design doc including the compatibility contract's frozen list. No golden changes, since the event value is runtime data rather than schema, and no migration, since neither the columns nor the index key move. `HydrationStart` and `set_hydration_start` keep their names. That event feeds both the `started` stage and `started_at` in the hydration times relation, and `started` says what it is: the dataflow was unsuspended. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
Two pieces of golden guidance were written on the retraction-delay branch, which is deferred and will not merge, so this branch still carried the claim CI disproved. `catalog_server_explain.slt` does need changing when a builtin log is added. Its `o.id NOT LIKE 'si%'` filter is real, so no new EXPLAIN entry appears, but the existing plans embed the inlined builtin `VALUES` sets as `Constant (N rows)` nodes, so every count over a catalog relation that gained a row moves. The question is not whether a plan is added but whether the existing plans change. Name `cluster.slt`'s index-column-position listing and `cockroach/srfs.slt`'s `unnest(indkey)` rows explicitly, since adding a column to an existing log reaches exactly those two and leaves the rest alone, and neither is found by searching for a count. Record the method that finds this class: read every file that mentions the relation by name, rather than reasoning about which kinds of value could have moved. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
`SUBSCRIBE ... UP TO x AS OF x` really did report `snapshot_complete` without computing the snapshot, and the comment claiming otherwise was wrong. The sink manufactures an empty upper once `up_to <= frontier`, and for equal bounds the frontier only has to reach x, leaving x itself uncomputed. Exclude an empty upper at that call site. The check belongs there and not in `observe_snapshot`, which is the distinction missed twice before. In `report_frontiers` an empty frontier comes from the dataflow's own progress reaching its end, so every time including the as-of is final and the stage is correct. In `process_subscribes` the empty upper is manufactured on completion and says nothing about the as-of. Rejecting empty in the shared helper is what previously left inputless collections permanently short of the stage. `ExportState::as_of` becomes `initial_as_of`, so the name carries what a paragraph of comment was carrying: a dataflow retained across reconciliation logs no new `Export` event, so the field keeps its install-time value. The `metric_sink` note was a third paragraph of defensive prose above one line of code. Two lines say the necessary thing: this is not `frontier_owner`, despite the same expression. Drop "grain" throughout the design doc in favour of saying which events are reported per worker and which per object, and rewrite the opening explanation of why timestamp columns cannot carry the lifecycle. Soften the compatibility contract's framing. "Frozen, and nothing will be removed" promises more than a design doc can. It now says these are intended to hold and that breaking one needs coordinating with consumers, which is the actual commitment. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
Main rewrote `metric_sink.rs` around the frontier report, and the merge left two adjacent comment paragraphs making overlapping points about the same line. Fold them into one. The `false` argument still holds after that rewrite: the shared frontier is written before the `worker_id != active_worker_id` early return, so every worker still writes it and no single copy carries progress. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
The rename to `snapshot_complete` replaced the exact token `` `hydrated` `` and so missed every other form the name appears in. Eight sites survived, including the SQL in "`installed` is the denominator", which the stable-contract list cross-references by name as the all-workers-reported test. `event = 'hydrated'` matches no row, so a consumer copying that query gets `0 = N`: a readiness gate that never fires and reports no error. The doc is what the PR points downstream rollups at, so that is the worst place for it. The others were compound backticked expressions the single-token replacement could not see: `write_unblocked - hydrated` in the negative-interval argument, `hydrated - started` and `write_unblocked - hydrated` in the per-interval cause list that the `write_unblocked` naming rests on, and `hydrated - started` in the justification for carrying the as-of in `details`. Two rustdoc leftovers in `compute_state.rs` and `logging/compute.rs` go with them, plus two prose descriptions of the log's own stages. Checking for zero remaining occurrences of the one token that had been replaced was not a check. The recheck now covers every quoting form the name can take. Part of CPU-226 Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
Review feedback: the prose argued its way to each decision rather than stating it. Each subsection now defines the thing it names. The `worker_id`, `dataflow_id`, `installed`, frontier, `write_blocked`, `write_unblocked` and `reason`/`details` subsections are cut to their definitions. `reason` and `details` now list the values a consumer can see rather than how to think about the split between them. Dropped: the implementation touch points and the settled-during-review list, both of which stop being useful once the change lands, the justification for keying by export, and a paragraph on refresh schedules distorting `hydration_time` that said nothing the section had not. Kept from the touch points, under its own heading, the two goldens that a column addition reaches and that a search for a count does not find. The two `reason` attributions that are not observable today move to Follow-up work, where the other follow-ups already live. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
Review feedback. Each comment now states what the thing is, not the reasoning that arrived at it. `LifecycleStage` says on each variant which event logs it, instead of a note above the enum, and drops the argument for why the stage is not named for hydration. `columns` drops the note on the vocabulary being closed, which its own signature says. `lifecycle_rows` keeps the reason the rows are retained and drops the prose on the relation's key, its memory cost and the at-most-once property. `lifecycle_details` is inlined into its one caller, which now takes the as-of and formats it through `make_string_datum` rather than allocating a `String` per event. `observe_snapshot` no longer explains what an empty antichain is. `observe_writes` keeps what `written` promises and drops the argument for it. `set_sink_write_frontier` states the invariant on `owned` rather than what breaks without it. `metric_sink` drops the note contrasting its election with the sink one. Also trimmed, in the same spirit: the frontier collection comment in `report_frontiers`, the empty-upper comment in `process_subscribes`, the `owns_sink_frontier` and `logged_stages` field docs, and the `drop_collection` note. The catalog description is one sentence. The ontology link keeps the clause naming the (export_id, worker_id) join, since the link itself declares only export_id, and drops the rest. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
The write stages are properties of the persist sink, so `mint` reports
them. It is the only place that tracks the output shard's upper, and it
runs on one elected worker, which is what makes the three events one
report per object rather than one per worker. The election therefore
never leaves the sink: `frontier_owner` is private, `owns_sink_frontier`
and `observe_writes` are gone, and `set_sink_write_frontier` takes only
the frontier it publishes for the controller's meet.
This relaxes the ordering between the two sides of the lifecycle, which
is what made the move possible. `mint` cannot see the dataflow's own
progress, so the write stages can no longer be gated on
`snapshot_complete`. For a shard that already holds data the as-of is
bounded below its upper, so `written` is true from installation, and
`apply_refresh` advances the upper of a `REFRESH` materialized view
before its dataflow computes anything. The compute stages stay ordered
among themselves, the write stages among themselves, and a consumer must
not read the six as one sequence. The contract and the invariant test say
so.
Two definitions get sharper on the way. `write_blocked` now means the
sink has a batch to mint and read-only mode forbids writing it, which is
the condition `maybe_mint_batch_description` already evaluates, rather
than the snapshot-completed proxy that stood in for it. `written` now
means the output is durable through the as-of, dropping the "and this
replica was permitted to write" conjunct that existed only to keep the
stages ordered.
`WriteStageLogger` in `crate::sink` holds the one copy of this logic for
both MV sink implementations. It reaches the logger through
`scope.worker().logger_for("materialize/compute")`, as the sink's `write`
operator already does for `CorrectionLogger`.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
`handle_schedule` needs a dataflow's other exports, and inferred them from `Rc` bookkeeping: a strong count of two meant "one export, plus the clone I just took", and anything else fell back to a pointer-equality scan over every collection. Both encoded a fact about the function's own locals rather than about the dataflow. `ComputeState::dataflow_exports` records the export set directly, keyed by dataflow index, which timely mints from a per-worker counter that only increases. `insert_collection` is now the one path that installs a collection, so the two maps cannot drift. That leaves one representation of the fact rather than two, so `dataflow_index` goes back to a plain `usize` and `drop_collection` decides by emptiness of the export set instead of by `Rc::try_unwrap`. The same map also answers whether the dataflow is still suspended, by asking whether any of its exports still holds a token, replacing a second `Rc::strong_count` read. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
Moving them into `mint` dropped the gate that kept `written` last, and the contract still claimed the write stages were ordered among themselves. They are not. `written` reads the output shard's upper, which moves whether or not this replica may write, and `write_blocked` additionally needs the dataflow's desired frontier to pass that upper. In the one case that produces `write_blocked` at all, a replacement awaiting a cutover, the shard already holds data, so the as-of sits below the upper and `written` is reported first. The contract now promises order only among the compute stages, and says a difference between two events is an elapsed interval only there. The testdrive comments say the same. The invariant that `write_unblocked` implies `write_blocked` stands, since the sink reports being unblocked only for a sink it reported blocked, but that is a causal dependency rather than an ordering. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz
fb01c00 to
fe9c7b4
Compare
|
Rebased onto Main had added a builtin log of its own in the meantime, so every id conflicted:
The replay was faithful: Also in this push, from the QA comment above: the write stages are ordered against nothing, and the contract said otherwise. One thing I did not change, deliberately.
Generated by Claude Code |
4 participants
Motivation
The compute half of improved hydration visibility needs to say, per maintained object, when it started to compute, when its snapshot was complete, and when its output became durable, so a query can take min/max to determine cluster replica readiness.
Timestamp columns cannot carry that. Some stages are per worker and others are not: each worker computes its own fragment of the dataflow, so
installed,startedandsnapshot_completehappen once per worker, while whether the output is durable is a property of the sink as a whole, maintained on one elected worker. And a timestamp cannot say why the next stage has not happened, so a NULL cannot tell a replacement materialized view awaiting a cutover apart from an index that will never write.Design doc:
doc/developer/design/20260817_compute_hydration_timestamps.md, reworked in this change. Part of CPU-226.Description
One append-only log relation,
mz_introspection.mz_compute_lifecycle_events_per_worker, per replica, in memory:reasoninstalledstartedsnapshot_completewrite_blockedread_onlywrite_unblockedwrittenAn index emits the first three and stops, which is the index degeneracy of the lifecycle falling out of the model rather than being special-cased. Subscribes and
COPY TOstop early for the same reason, and a metric sink folds its output into the metrics registry rather than into a shard, so it has no write stages either.mz_compute_hydration_times_per_workeris untouched. All six of its columns still report exactly what they reported before, somz_compute_hydration_times,mz_compute_hydration_statuses,mz_hydration_statusesand the blue-green readiness query are unaffected. This change is additive to them.The two readings share no vocabulary, deliberately.
hydrated_atis the durability reading, taken when the reported output frontier passes the as-of, and that frontier is the meet of the write and compute frontiers. For a collection that sinks to persist it moves only once the output is durable; for an index, which produces its output by writing its own trace, it coincides with computation. One column therefore meant two different things depending on the object. The log splits that intosnapshot_complete, always the dataflow-progress reading, andwritten, always the durability one, neither varying by object type. The stage is not namedhydratedfor exactly that reason.Which worker reports what.
worker_idis the worker that observed the event and is never NULL. The per-object events are observed by the single worker that maintains the sink's shared write frontier, so they appear once per object and the row records which worker was elected. Nothing else may read that shared frontier as a measure of writing:mintclears it on every non-elected worker, where it is the empty antichain and would report having written everything immediately. The electionhashed(sink_id) % peershas one definition,crate::sink::frontier_owner, called both bymintand by the code recording ownership.dataflow_id.installedandstarteddescribe the dataflow, which can maintain more than one export, whilesnapshot_completeand the write stages describe one export. Keying the relation by export keeps every row answerable bymz_objects.id, and carrying the dataflow id makes the shared events recognizable as shared:SELECT DISTINCT dataflow_id, event, occurred_atrecovers the dataflow-level facts. A dataflow carries the same index on every worker of a replica.startedis reported when the dataflow actually unsuspends. A dataflow's exports share one suspension token, so computation begins only once every export has been scheduled. Reporting per export from its ownSchedulewould date the earlier ones to before their dataflow was running. This also fixesmz_compute_hydration_times_per_worker.started_at, latently, since nothing in production ships more than one export per dataflow.The write stages are gated on
snapshot_complete. Before it the sink has produced nothing, so read-only mode is holding nothing back, and every collection starts read-only, so reporting a block from installation would put awrite_blockedandwrite_unblockedon essentially every materialized view, both ahead of the snapshot. Gating also keepswrittenordered after it:apply_refreshrounds aREFRESHmaterialized view's frontier up off its input frontier, before the dataflow computes anything, so the shard's upper passes the as-of while the dataflow is still computing. That also means a refresh schedule advances writing rather than blocking it, so there is norefreshcause forwrite_blockedand that value is not in the vocabulary.What
writtendoes and does not promise. It says the output is durable through the as-of and this replica was permitted to write. It does not say this replica performed the write, and cannot: every replica'smintreads the shard's upper back from persist, so the frontier advances on all of them when any one wins the append. Replicas of a cluster produce identical batches and race to append, so which won is not operationally meaningful, but it does meanwrittenlands withsnapshot_completeon a restarted or scaled-out replica and withwrite_unblockedat a cutover. Attribution needs a signal from the replica's own append path, which has to cross workers, and is recorded as follow-up.Only
read_onlyis attributed. It is the one cause of a write block compute can observe. Two further attributions are not available and are called out as follow-up: distinguishing astartedthat waited on the hydration limiter from one that waited on its inputs needsSequentialHydrationto report which, and distinguishing a freshCreateDataflowfrom a dataflow retained across reconciliation is not observable here at all.reasonis typed,detailsis not, followingmz_source_statusesandmz_sink_statuses.detailscarries the dataflow's as-of, without which the interval between two stages says nothing about how much work was done.The relation is unkeyed, so
index_byarranges by the whole row.(export_id, worker_id, event)is unique today, but declaring it a key is a uniqueness claim the optimizer acts on, and a false key is a correctness hazard where a wide index is only a performance one.A compatibility contract is in the design doc under "What the timestamps promise, and what they do not", because downstream rollups are being built on these relations now. Intended to hold, with a break treated as needing coordination rather than as a detail: the six event values and their meanings, which events are per worker and which per object,
installedas the all-workers-reported denominator,occurred_atas a wallclock instant carrying its worker's anchor,snapshot_completeandwrittennever varying by object type,hydrated_at/time_nsstaying the demux-computed durability reading rather than becoming a view, and(export_id, worker_id)as the exact join between the two relations. Open sets a consumer must tolerate: neweventvalues, newreasonvalues, newdetailskeys. Notably, rows are retracted when the replica processes the drop, which is asynchronous with the catalog transaction, so a consumer will see rows whoseexport_idhas leftmz_objectsand must not filter with an inner join.Verification
cargo check --all-targetsis clean formz-compute,mz-compute-client,mz-catalogandmz-sqllogictest— the last becausemz_environmentd::Confighas a constructor in that crate which-p mz-environmentddoes not cover.bin/fmtpasses.test/testdrive/compute-lifecycle-events.tdis new. It asserts the per-worker stage counts for an index and the per-object count for a materialized view'swritten, the ordering within each worker,writtennever precedingsnapshot_complete,occurred_atlanding in the recent past,details->>'as_of'being present,dataflow_idagreeing withmz_compute_exports_per_workerfor the same export and worker, retraction on drop, and three invariants underset-max-tries max-tries=1: the closedeventandreasonvocabularies, no stage without its predecessors, and no stage reported twice per export and worker.Limitations.
write_blockedandwrite_unblockedare not covered, because reaching them needs a replica still read-only after a snapshot has completed, which is a 0dt cutover and not reachable from plain testdrive. The invariant assertions are written so the pair may be absent or complete, never requiring it to be present. A platform-check or cluster test driving read-only mode would be the place to cover them.Goldens were hand-edited, since sqllogictest cannot be built in the environment this was developed in. Two rounds of CI corrections have been folded in, and the design doc now names the two that a column addition reaches and that a name search finds but a count search does not:
cluster.slt's index-column-position listing andcockroach/srfs.slt'sunnest(indkey)rows.Review rounds folded in
Four automated QA passes, one adversarial pass, and human review.
Human review found a real bug that an earlier version of this description wrongly claimed was already fixed:
SUBSCRIBE ... UP TO x AS OF xreportedsnapshot_completewithout computing the snapshot. The sink manufactures an empty batch upper onceup_to <= frontier, and with equal bounds the frontier only has to reach x, leaving x uncomputed. Now fixed by excluding an empty upper at that call site specifically, not in the shared helper: inreport_frontiersan empty frontier means the dataflow's own progress reached its end, so the stage is correct there, and rejecting empty in the helper is what previously left inputless collections permanently short of the stage.Also fixed across the rounds: an attempted write baseline that latched
mint's placeholder frontier, reverted in favour of documenting whatwrittenpromises; at-most-once made structural by keying the retained rows by stage;frontier_ownergiven one home incrate::sink, with a note on whymetric_sink's identical election stays separate; a paired setter forsink_write_frontierandowns_sink_frontier; a quadratic collection scan inhandle_schedule; a falseManyToOneontology link ondataflow_id, whose target holds one row per rendered object; and a contract line wrongly promising that a lifecycle row implies a live export.Declined and documented:
lifecycle_rowsretains packedRows rather than re-deriving them at drop time, which costs about a kilobyte per export per worker and buys a retraction that is correct by construction.