compute: drop the duplicated instants from the hydration times relation

`installed_at` and `started_at` were maintained twice, in this relation and in
`mz_compute_lifecycle_events_per_worker`, from the same handlers and the same
event times, with the `installed_at <= started_at` invariant enforced in both
representations. Keep them in one place: the lifecycle log's `installed` and
`started` events.

`time_ns` and `hydrated_at` stay. They are not duplicates of anything, because
`hydrated_at` is the durability reading, taken when the reported output frontier
passes the as-of. That frontier is the meet of the write and compute frontiers, so
for a collection that sinks to persist it moves only once the output is durable,
while for an index, which produces its output by writing its own trace, it
coincides with computation. One column meant two things depending on the object.

The lifecycle log answers that by using separate terms rather than redefining the
column: its `hydrated` is always the dataflow-progress reading and its `written`
is always the durability one, neither depending on the object type. Pivoting the
log's `hydrated` into `hydrated_at` would therefore not be a refactor but a
redefinition, and since `mz_compute_hydration_statuses.hydrated` is `time_ns IS
NOT NULL` and feeds the blue-green readiness query, it would have readiness cut
over on the earlier compute reading, before the output is durable. Comments at
both relations record which reading each term carries.

`handle_hydration_start` no longer writes to the hydration-time output, since
nothing it changes is packed there any more; its retract-and-reinsert pair would
have been a no-op. `started_at` survives as internal state only, to guard the
handler and to back-fill `started` in `handle_hydration`.

`test/testdrive/hydration-timestamps.td` moves its stage scenarios to the
lifecycle relation and keeps the duration assertions here, so it now tests that
the two relations agree rather than that one of them is self-consistent. The
`time_ns` cross-check joins them on `(export_id, worker_id)`, both being per
worker.

No new migration step: the index key `(export_id, worker_id)` is unchanged, and
the `mz_indexes` fingerprint move is covered by the existing replacement step at
this dev version.

Part of CPU-226

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz

Author: claude

doc/developer/design/20260817_compute_hydration_timestamps.md

@@ -236,6 +236,28 @@ event, occurred_at` recovers the dataflow-level facts, and the redundancy is vis
 rather than implied. The column also means the relation carries the export-to-dataflow
 mapping that `mz_compute_exports_per_worker` holds, for eight bytes a row.
 
+**One reading per term.** `mz_compute_hydration_times_per_worker.hydrated_at` is the
+durability reading, taken when the reported output frontier passes the as-of. That
+frontier is the meet of the write and compute frontiers, so for a collection that
+sinks to persist it moves only once the output is durable, while 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 type, and a
+consumer could not tell which without knowing what it was looking at.
+
+The lifecycle log splits that ambiguity into separate terms rather than redefining the
+column: `hydrated` is always the dataflow-progress reading and `written` is always the
+durability one, neither depending on the object type. `hydrated_at` and `time_ns` stay
+where they are, because `mz_compute_hydration_statuses` and the blue-green readiness
+query are defined on them, and moving readiness onto the earlier compute reading would
+have it cut over before the output is durable.
+
+What does move is `installed_at` and `started_at`. Those were unambiguous, so keeping
+them in both relations meant two mechanisms maintaining the same instants and the same
+`installed_at <= started_at` invariant. They are now the lifecycle log's `installed`
+and `started` events alone. The two relations join on `(export_id, worker_id)`, both
+being per worker, so a consumer that wants the queueing interval and the duration
+together reads them from one join rather than from one row.
+
 **`installed` is the denominator.** A consumer asking whether every worker has reached
 a stage cannot count rows and compare against a worker count it does not have, and it
 should not have to join the catalog to find one. The relation is append-only, so a

src/compute-client/src/logging.rs

@@ -353,18 +353,24 @@ impl LogVariant {
                 .with_key(vec![0, 1])
                 .finish(),
 
+            // NOTE: `hydrated_at` here is the *durability* reading, taken when the reported
+            // output frontier passes the as-of. That frontier is the meet of the write and
+            // compute frontiers, so for a collection that sinks to persist it moves only once
+            // the output is durable, while for an index, which produces its output by writing
+            // its own trace, it coincides with computation. One column therefore meant two
+            // things depending on the object.
+            //
+            // `mz_compute_lifecycle_events_per_worker` splits that ambiguity into separate
+            // terms: its `hydrated` is the dataflow-progress reading and its `written` is the
+            // durability one, so neither depends on the object type. This relation keeps the
+            // combined reading, because `mz_compute_hydration_statuses` and the blue-green
+            // readiness query are defined on it. The instants that were unambiguous,
+            // `installed_at` and `started_at`, moved to the lifecycle relation rather than
+            // being maintained in both.
             LogVariant::Compute(ComputeLog::HydrationTime) => RelationDesc::builder()
                 .with_column("export_id", SqlScalarType::String.nullable(false))
                 .with_column("worker_id", SqlScalarType::UInt64.nullable(false))
                 .with_column("time_ns", SqlScalarType::UInt64.nullable(true))
-                .with_column(
-                    "installed_at",
-                    SqlScalarType::TimestampTz { precision: None }.nullable(false),
-                )
-                .with_column(
-                    "started_at",
-                    SqlScalarType::TimestampTz { precision: None }.nullable(true),
-                )
                 .with_column(
                     "hydrated_at",
                     SqlScalarType::TimestampTz { precision: None }.nullable(true),

src/compute/src/logging/compute.rs

@@ -735,10 +735,6 @@ impl DemuxState {
             make_string_datum(export_id, &mut self.scratch_string_a),
             Datum::UInt64(u64::cast_from(self.worker_id)),
             Datum::from(time_ns),
-            epoch_offset_datum(timestamps.installed_at),
-            timestamps
-                .started_at
-                .map_or(Datum::Null, epoch_offset_datum),
             timestamps
                 .hydrated_at
                 .map_or(Datum::Null, epoch_offset_datum),
@@ -861,6 +857,10 @@ impl DemuxState {
 
 /// Wallclock instants of an export's hydration lifecycle, as durations since the Unix epoch.
 ///
+/// Only `hydrated_at` reaches `mz_compute_hydration_times_per_worker`. `installed_at` and
+/// `started_at` are kept to order and back-fill the lifecycle log's `installed` and `started`
+/// events, which are where a consumer reads them.
+///
 /// These are sampled from compute logging event times, which advance off an `Instant` anchored to
 /// the epoch once per worker when logging is initialized. They are therefore monotone within a
 /// worker and unaffected by system clock steps, but they carry that worker's anchor, so comparing
@@ -1184,8 +1184,7 @@ impl DemuxHandler<'_, '_, '_> {
         &mut self,
         HydrationStartReference { export_id }: Ref<'_, HydrationStart>,
     ) {
-        let ts = self.ts();
-        // Stamp the event time rather than `ts`, as in `handle_export`.
+        // Stamp the event time rather than the rounded update timestamp, as in `handle_export`.
         let started_at = self.time;
         let export_id = Columnar::into_owned(export_id);
 
@@ -1200,22 +1199,12 @@ impl DemuxHandler<'_, '_, '_> {
             return;
         }
 
-        let old_timestamps = export.hydration_timestamps;
         export.hydration_timestamps.started_at = Some(started_at);
-        let new_timestamps = export.hydration_timestamps;
-        let time_ns = export.hydration_time_ns;
-
-        let retraction = self
-            .state
-            .pack_hydration_time_update(export_id, time_ns, &old_timestamps);
-        self.output
-            .hydration_time
-            .give((retraction, ts, Diff::MINUS_ONE));
-        let insertion = self
-            .state
-            .pack_hydration_time_update(export_id, time_ns, &new_timestamps);
-        self.output.hydration_time.give((insertion, ts, Diff::ONE));
 
+        // No `hydration_time` output here. That relation carries only `time_ns` and
+        // `hydrated_at`, neither of which this event changes, so the retract-and-reinsert pair
+        // would be a no-op. `started_at` is now kept only to guard this handler and to
+        // back-fill `started` in `handle_hydration`.
         self.log_lifecycle(export_id, LifecycleStage::Started);
     }
 

test/sqllogictest/mz_catalog_server_index_accounting.slt

@@ -358,8 +358,6 @@ mz_compute_hydration_times  replica_id
 mz_compute_hydration_times  time_ns
 mz_compute_hydration_times_per_worker  export_id
 mz_compute_hydration_times_per_worker  hydrated_at
-mz_compute_hydration_times_per_worker  installed_at
-mz_compute_hydration_times_per_worker  started_at
 mz_compute_hydration_times_per_worker  time_ns
 mz_compute_hydration_times_per_worker  worker_id
 mz_compute_import_frontiers_per_worker  export_id