|
| 1 | +# Durable Object Hydration History |
| 2 | + |
| 3 | +## Context |
| 4 | + |
| 5 | +Materialize exposes current hydration state, but that state disappears when a |
| 6 | +dataflow or a replica restarts. A user can tell whether an object is hydrated |
| 7 | +now, but not how long the last hydration took, or whether today's is unusually |
| 8 | +slow. |
| 9 | + |
| 10 | +This records completed compute-object hydration episodes in a durable table. It |
| 11 | +is the first stage of broader hydration history. Replica-wide episodes, resource |
| 12 | +peaks, failed episodes, and storage objects need signals that do not exist yet. |
| 13 | + |
| 14 | +## Goals |
| 15 | + |
| 16 | +- Record successful hydration of indexes and materialized views per replica. |
| 17 | +- Survive environmentd and replica restarts. |
| 18 | +- Stay idempotent across concurrent environmentd processes. |
| 19 | +- Leave the values of existing hydration relations untouched. |
| 20 | +- Bound storage with configurable retention. |
| 21 | +- Default off in production, on in CI. |
| 22 | + |
| 23 | +## Non-Goals |
| 24 | + |
| 25 | +Each of these is excluded because the signal to do it correctly is missing, not |
| 26 | +because it is unwanted: |
| 27 | + |
| 28 | +- **Failed or canceled episodes.** A replica cannot report its own crash, and the |
| 29 | + current-state log retracts an object without recording why. |
| 30 | +- **Replica-wide episodes.** Correct boundaries need the object set present at the |
| 31 | + transition from hydrated to hydrating, and explicit transitions between those |
| 32 | + states. |
| 33 | +- **Resource peaks.** The metrics history holds samples, not high-water values. |
| 34 | + Presenting samples as peaks would mislead. |
| 35 | +- **Sources and sinks.** Storage publishes no equivalent lifecycle timestamps. |
| 36 | +- **Unmanaged replicas.** Their worker count is unknown, and without it a |
| 37 | + cross-worker aggregate cannot be shown to be complete. |
| 38 | + |
| 39 | +## Design |
| 40 | + |
| 41 | +Compute stamps three replica-side timestamps per export and worker: `installed_at` |
| 42 | +when the dataflow is installed, `started_at` when it is unsuspended, and |
| 43 | +`hydrated_at` when the output frontier passes the as-of. They are added to the |
| 44 | +existing `mz_introspection.mz_compute_hydration_times_per_worker` log, which keeps |
| 45 | +its name, OID, and object kind, so its generated per-replica index and every |
| 46 | +relation built on it are unaffected. A consumer doing `SELECT *` sees three new |
| 47 | +columns. |
| 48 | + |
| 49 | +Renaming the log and leaving a compatibility view behind was considered, since the |
| 50 | +compute half of this project proposes it. Not done here. It would move OID 16977 to |
| 51 | +a differently shaped relation and flip the old name from a log to a view. Naming |
| 52 | +that relation is the compute team's call on their own change. |
| 53 | + |
| 54 | +A coordinator task visits one managed user replica per interval. It installs an |
| 55 | +internal subscribe on that replica which aggregates complete worker rows, maps |
| 56 | +runtime export ids to catalog objects, anti-joins against the history table, and |
| 57 | +writes the missing rows through the timestamped OCC read-then-write path. |
| 58 | + |
| 59 | +## History Table |
| 60 | + |
| 61 | +```text |
| 62 | +mz_internal.mz_object_hydration_history |
| 63 | + object_id text not null |
| 64 | + cluster_id text not null |
| 65 | + replica_id text not null |
| 66 | + installed_at timestamptz not null |
| 67 | + started_at timestamptz null |
| 68 | + finished_at timestamptz null |
| 69 | + status text not null |
| 70 | + key (object_id, replica_id, installed_at) |
| 71 | +``` |
| 72 | + |
| 73 | +`installed_at` is part of the episode identity because it is replica-stamped and |
| 74 | +stable across an environmentd restart. `started_at` cannot serve that role, it is |
| 75 | +null while an object waits to run. |
| 76 | + |
| 77 | +Only terminal `hydrated` rows are written today, so every row has a finish time. |
| 78 | +The nullable columns and the `status` column reserve a compatible shape for |
| 79 | +episodes that are canceled or unfinished, once those become observable. Retention |
| 80 | +keys off `finished_at`, so an unfinished episode needs a second age basis before it |
| 81 | +can be recorded at all. |
| 82 | + |
| 83 | +The table is in `mz_internal`, not `mz_catalog`, because its contents are best |
| 84 | +effort and `status` will gain values. `mz_internal` carries no stability |
| 85 | +commitment. Its closest sibling, `mz_internal.mz_object_arrangement_size_history`, |
| 86 | +sits there for the same reason. |
| 87 | + |
| 88 | +One index on `object_id`, on the catalog server, serves the question users ask: |
| 89 | +how long did this object take to hydrate. The collector's anti-join has the same |
| 90 | +shape but cannot use it, because that subscribe runs on the targeted *user* |
| 91 | +replica where the index does not exist. Retention scans by `finished_at` and is |
| 92 | +deliberately left unindexed. It runs once per sweep over a table bounded by |
| 93 | +retention, and a second arrangement costs every environment memory. |
| 94 | + |
| 95 | +## Why concurrent writers converge |
| 96 | + |
| 97 | +The subscribe reads the history table it is about to write. That is what makes the |
| 98 | +write idempotent across processes, and it is the reason the read expression looks |
| 99 | +redundant. |
| 100 | + |
| 101 | +Two environmentd processes computing the same missing row at frontier `T` both |
| 102 | +submit a timestamped write for `T`. One commits and advances the table upper. The |
| 103 | +other is told the timestamp passed, waits for its own subscribe to progress, sees |
| 104 | +the committed row appear, and its anti-join retracts the candidate. There is |
| 105 | +nothing left to write. |
| 106 | + |
| 107 | +The losing writer never retries stale diffs at the next eligible timestamp. It |
| 108 | +only retries with state it has observed to be valid at the frontier it is writing |
| 109 | +at. That property is also the source of the frontier-skew limitation below. |
| 110 | + |
| 111 | +## What the aggregation has to guard against |
| 112 | + |
| 113 | +Three conditions in the collection query each prevent a specific wrong row. None |
| 114 | +of them are obvious from reading the query. |
| 115 | + |
| 116 | +**Group by dataflow, not by catalog item.** A materialized view can own two |
| 117 | +`GlobalId`s at once, because a replacement installs the new dataflow's id while the |
| 118 | +old one still serves reads. Grouping by item mixes both dataflows' worker rows, |
| 119 | +presents twice the expected worker count, and records neither episode. |
| 120 | + |
| 121 | +**Require `max(installed_at) <= min(hydrated_at)`.** A replica whose processes |
| 122 | +restarted at different times can otherwise present worker 0 from the old process |
| 123 | +and worker 1 from the new one, satisfy the worker-count check, and yield one row |
| 124 | +whose duration silently includes the downtime between two episodes. |
| 125 | + |
| 126 | +**Report `started_at` only when every worker observed one.** A partially observed |
| 127 | +start is not the episode's start. An import-free dataflow is never suspended, so |
| 128 | +its `Schedule` can arrive after it has already hydrated, and recording that late |
| 129 | +arrival would invent an interval nobody measured. |
| 130 | + |
| 131 | +In practice index exports report a start and materialized view exports frequently |
| 132 | +do not, so `started_at` is optional for consumers. What always holds is |
| 133 | +`installed_at <= finished_at`, and `installed_at <= started_at <= finished_at` for |
| 134 | +a non-null start. Why materialized view exports so often observe no start is worth |
| 135 | +running down before `started_at` is presented as a queueing signal. |
| 136 | + |
| 137 | +An episode's interval spans workers, using the minimum installation and start and |
| 138 | +the maximum completion, so it still includes clock skew between worker processes. |
| 139 | + |
| 140 | +Multi-export dataflows would need one more step. All exports of a dataflow share a |
| 141 | +suspension token, so the dataflow starts only once every export is scheduled, while |
| 142 | +the stamp is per export. Every compute dataflow has exactly one export today and |
| 143 | +`sequential_hydration.rs` asserts it, so the two coincide. |
| 144 | + |
| 145 | +## What is and is not recorded |
| 146 | + |
| 147 | +Collection samples current state. It is not an event log, and the log it reads |
| 148 | +retracts an object's row when the export goes away. An episode is recorded only if |
| 149 | +its row is still live when its replica's turn comes around, so an object dropped |
| 150 | +before then, or a replica process that restarts before then, leaves no trace. An |
| 151 | +object that never reports a completion time, such as a constant materialized view |
| 152 | +whose frontier jumps straight to empty, is never recorded. |
| 153 | +`test/testdrive/hydration-status.td` asserts that absence by name, so the limit is |
| 154 | +pinned rather than merely tolerated. |
| 155 | + |
| 156 | +Making these durable requires compute to emit hydration transitions into an |
| 157 | +append-only collection that survives until an observer acknowledges them. Until |
| 158 | +then, best effort is the honest description of a sampler. |
| 159 | + |
| 160 | +## Retention |
| 161 | + |
| 162 | +`hydration_history_retention_period` defaults to 30 days. Retention is another OCC |
| 163 | +mutation: it subscribes to rows older than the cutoff and writes their retractions |
| 164 | +at the observed frontier. Collection applies the same cutoff, so a still-live log |
| 165 | +row cannot resurrect an episode retention just retracted. |
| 166 | + |
| 167 | +Retention deletes a bounded batch per sweep and converges over as many sweeps as it |
| 168 | +takes. The bound is not a nicety. The OCC path refuses a selection larger than |
| 169 | +`max_result_size` before submitting any write, so one unbounded delete over a large |
| 170 | +backlog would fail identically forever and never shrink the table. The bound has to |
| 171 | +sit inside a derived table, because a top-level `LIMIT` lands in the plan's |
| 172 | +`RowSetFinishing`, which the OCC path deliberately discards, and the delete would |
| 173 | +be silently unbounded again. |
| 174 | + |
| 175 | +Retention runs on the catalog server, so it keeps working when there are no user |
| 176 | +replicas at all, and it runs even when that sweep's collection failed. A |
| 177 | +crash-looping replica must not be able to stop the table from shrinking. The |
| 178 | +dependency does not run the other way: a catalog server without a replica skips |
| 179 | +retention and leaves collection running. |
| 180 | + |
| 181 | +Disabling collection also suspends retention. The alternative is an always-on |
| 182 | +subscribe in the default configuration, where the table is empty and there is |
| 183 | +nothing to retain. Rows already collected are therefore kept while collection is |
| 184 | +off. |
| 185 | + |
| 186 | +## Durability is best effort |
| 187 | + |
| 188 | +Builtin tables are reset at bootstrap and re-shard on a forced schema migration. |
| 189 | +The history table is exempt from both, because a sampled history cannot be rebuilt |
| 190 | +from anything else once it is gone. Those two exemptions are the whole promise. |
| 191 | + |
| 192 | +We do not commit to carrying the contents across every future version. A schema |
| 193 | +change to a builtin table allocates a fresh shard, so changing these columns clears |
| 194 | +the history. That is an acceptable trade: the value here is the distribution the |
| 195 | +table accumulates, and starting it over costs one retention period, while freezing |
| 196 | +the schema to protect it costs every later improvement. |
| 197 | + |
| 198 | +So we try not to break it and we do not promise not to. The exemption exists to |
| 199 | +stop an incidental migration from wiping the table as a side effect of unrelated |
| 200 | +work, not to make it untouchable. Giving it up is deliberate: an assert in |
| 201 | +`plan_migration` fires if a migration step ever names this table, and the comment |
| 202 | +there says to remove the assert and the exemption together and to note in the |
| 203 | +release notes that the history restarts. The user-facing documentation says the |
| 204 | +matching thing, that this is a record to look at and not a data source to build on. |
| 205 | + |
| 206 | +## Scheduling and isolation |
| 207 | + |
| 208 | +`hydration_history_collection_interval` sets the cadence and disables collection at |
| 209 | +zero. Sweeps never overlap, which bounds compute load and keeps the collector from |
| 210 | +contending with itself in the serialized timestamped-write path. |
| 211 | + |
| 212 | +Fires align to interval boundaries, and each sleep is capped, so that lowering a |
| 213 | +long interval at runtime takes effect within the cap rather than after the old |
| 214 | +interval elapses. Tests depend on that. A disabled collector polls far more |
| 215 | +coarsely, since that is the cadence of every environment in the default |
| 216 | +configuration. |
| 217 | + |
| 218 | +Two isolation decisions are worth stating outright: |
| 219 | + |
| 220 | +**Background mutations take no OCC write permit.** A session's wait for a permit is |
| 221 | +bounded by its statement timeout. A sweep has no statement timeout, and its |
| 222 | +subscribe must first hydrate a dataflow on a user replica, so holding a permit |
| 223 | +would let a background sampler stall user DML for as long as that takes. The sweep |
| 224 | +is single-flight, so skipping the permit adds at most one concurrent |
| 225 | +read-then-write. |
| 226 | + |
| 227 | +**The sweep is aborted when the coordinator is dropped.** Unlike a session it holds |
| 228 | +no `Client`, so nothing otherwise stops it from outliving the coordinator, and the |
| 229 | +runtime teardown that follows drops the timestamp oracle's worker task. A sweep |
| 230 | +still running then reads a timestamp from a dead oracle and panics. The coordinator |
| 231 | +owns the task handle so that dropping it cancels the sweep first. |
| 232 | + |
| 233 | +Each mutation has its own deliberately generous timeout. Even a mutation with |
| 234 | +nothing to write waits for its read to linearize, which can take a full |
| 235 | +`default_timestamp_interval`, a parameter with no upper bound, so a tighter bound |
| 236 | +would let a large timestamp interval starve retention permanently. |
| 237 | + |
| 238 | +Replica drop, cluster drop, dependency replacement, replica failure, and timeout all |
| 239 | +skip the attempt, and a later sweep recomputes from current state. Read-only |
| 240 | +generations do nothing. Replicas with introspection disabled are skipped, since |
| 241 | +their log arrangements exist but are never populated, so a subscribe would read a |
| 242 | +sealed, empty collection on every sweep forever. |
| 243 | + |
| 244 | +## Known limitation: frontier skew |
| 245 | + |
| 246 | +The write timestamp is the subscribe's observed frontier rather than a fresh oracle |
| 247 | +timestamp, for the reason given above. That frontier is the minimum over the |
| 248 | +subscribe's inputs, and one input is a replica-local introspection log whose |
| 249 | +frontier the *replica* advances from its own clock, rounded up to its introspection |
| 250 | +interval. |
| 251 | + |
| 252 | +So a replica whose clock trails environmentd by more than one introspection |
| 253 | +interval produces a frontier that never gets ahead of the timestamp oracle. Every |
| 254 | +attempt loses its timestamp race and the step exhausts its retries. The consequence |
| 255 | +is bounded. Nothing wrong is recorded, that replica records nothing at all, and the |
| 256 | +sweep stretches while it retries, delaying others in the rotation. It resolves |
| 257 | +itself when the clocks converge. |
| 258 | + |
| 259 | +Accepted for now: skew between processes is normally milliseconds while the |
| 260 | +tolerance is a whole introspection interval, and the failure mode is silent retry |
| 261 | +rather than wrong data. Because the symptom is otherwise hard to attribute, |
| 262 | +exhausting retries logs that the replica's introspection frontier may be trailing. |
| 263 | +Removing the limitation means deriving the write timestamp from the target table's |
| 264 | +own frontier, which needs a correctness argument for applying replica-derived diffs |
| 265 | +at a timestamp the replica never observed. |
| 266 | + |
| 267 | +## Notes on the catalog plumbing |
| 268 | + |
| 269 | +Two things here surprise on first contact and cost a CI round trip each. |
| 270 | + |
| 271 | +Adding a builtin index changes the fingerprint of `mz_catalog.mz_indexes`, which |
| 272 | +inlines the builtin index set as a VALUES list precisely so that this happens. The |
| 273 | +fingerprint check then panics at catalog open unless an explicit |
| 274 | +`MigrationStep::replacement` for `mz_indexes` is declared at the current dev |
| 275 | +version. This is deliberate, it guarantees stale data is never served, and it means |
| 276 | +adding an index is never only an index. |
| 277 | + |
| 278 | +The `test/sqllogictest/autogenerated/*.slt` goldens are generated from the |
| 279 | +user-facing docs markdown, but the sqllogictest run compares them against the live |
| 280 | +catalog. Editing a column comment in the markdown without editing it in the Rust |
| 281 | +builtin definition produces a golden that passes the docs lint locally and fails in |
| 282 | +CI. |
| 283 | + |
| 284 | +## Rollout |
| 285 | + |
| 286 | +The interval is zero, and so disabled, in production, and runtime configuration can |
| 287 | +enable it without restarting environmentd. The mzcompose configuration enables it at |
| 288 | +60 seconds so hydration, restart, retention, and catalog tests exercise the path. |
| 289 | + |
| 290 | +It stays off in the sqllogictest runner defaults, against the usual preference for |
| 291 | +enabling new paths in tests. The collector installs subscribes and writes a builtin |
| 292 | +table, while those runs assert on catalog contents and plans, so enabling it risks |
| 293 | +churn and timing flakiness in files that have nothing to do with hydration. |
| 294 | + |
| 295 | +## Future Work |
| 296 | + |
| 297 | +- Record installation and start before completion, then finalize canceled and |
| 298 | + failed episodes by joining replica lifecycle events. |
| 299 | +- Define a replica episode state machine from the object set present at the |
| 300 | + hydrated-to-hydrating transition. |
| 301 | +- Publish resettable per-process high-water values for RAM, swap, and scratch disk, |
| 302 | + and define replica aggregation without pretending sampled maxima are simultaneous |
| 303 | + peaks. |
| 304 | +- Give storage objects equivalent lifecycle timestamps. |
| 305 | +- Build replica history and progress views on those signals. |
0 commit comments