Skip to content

Commit be177e3

Browse files
committed
doc: design for durable object hydration history
Records completed compute-object hydration episodes in a durable table so that a hydration can be compared against previous ones after a restart. The document covers the collector's idempotence argument across concurrent environmentd processes, the aggregation conditions that each prevent a specific wrong row, the best-effort durability position, and the frontier-skew limitation the chosen write timestamp implies. Closes: SQL-632
1 parent 91d114e commit be177e3

1 file changed

Lines changed: 313 additions & 0 deletions

File tree

Lines changed: 313 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,313 @@
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+
No index. An arrangement on the catalog server would hold the whole table, which
89+
grows with objects times replicas times re-hydrations, and nothing queries this
90+
table by key yet. A user query instead scans a table bounded by retention. Note
91+
that an index would not help the collector anyway, since its anti-join runs on the
92+
targeted user replica, where a catalog-server arrangement does not exist.
93+
94+
For the same reason the table is not a retained-metrics object. That flag pins a
95+
30 day logical compaction window, which keeps 30 days of update history rather
96+
than current state. Our history is in the rows, which retention retracts on its
97+
own schedule, and nothing reads this table at an old timestamp.
98+
99+
## Why concurrent writers converge
100+
101+
The subscribe reads the history table it is about to write. That is what makes the
102+
write idempotent across processes, and it is the reason the read expression looks
103+
redundant.
104+
105+
Two environmentd processes computing the same missing row at frontier `T` both
106+
submit a timestamped write for `T`. One commits and advances the table upper. The
107+
other is told the timestamp passed, waits for its own subscribe to progress, sees
108+
the committed row appear, and its anti-join retracts the candidate. There is
109+
nothing left to write.
110+
111+
The losing writer never retries stale diffs at the next eligible timestamp. It
112+
only retries with state it has observed to be valid at the frontier it is writing
113+
at. That property is also the source of the frontier-skew limitation below.
114+
115+
## What the aggregation has to guard against
116+
117+
Three conditions in the collection query each prevent a specific wrong row. None
118+
of them are obvious from reading the query.
119+
120+
**Group by dataflow, not by catalog item.** A materialized view can own two
121+
`GlobalId`s at once, because a replacement installs the new dataflow's id while the
122+
old one still serves reads. Grouping by item mixes both dataflows' worker rows,
123+
presents twice the expected worker count, and records neither episode.
124+
125+
**Bound how far apart the per-worker installation stamps are.** A replica can
126+
otherwise present worker 0 from before a re-install and worker 1 from after it,
127+
satisfy the worker-count check, and yield one row whose duration silently spans
128+
two episodes.
129+
130+
The bound compares installations to each other rather than an installation
131+
against a completion. Each process anchors its logging clock at its own
132+
`SystemTime`, so a cross-field comparison measures clock skew as much as elapsed
133+
work. On a replica with more than one process, skew larger than the hydration
134+
itself would reject a complete episode, and would keep rejecting it, because the
135+
log values never change. That failure would be invisible, since the object reports
136+
hydrated everywhere else while the history stays empty, and it would only appear
137+
on the large replicas this feature exists for.
138+
139+
**Report `started_at` only when every worker observed one.** A partially observed
140+
start is not the episode's start. An import-free dataflow is never suspended, so
141+
its `Schedule` can arrive after it has already hydrated, and recording that late
142+
arrival would invent an interval nobody measured.
143+
144+
In practice index exports report a start and materialized view exports frequently
145+
do not, so `started_at` is optional for consumers. What always holds is
146+
`installed_at <= finished_at`, and `installed_at <= started_at <= finished_at` for
147+
a non-null start. Why materialized view exports so often observe no start is worth
148+
running down before `started_at` is presented as a queueing signal.
149+
150+
An episode's interval spans workers, using the minimum installation and start and
151+
the maximum completion, so it is inflated by clock skew between processes. That is
152+
bounded by the skew itself, which is small next to a hydration worth recording. An
153+
install generation carried in the log would remove the wall-clock dependency for
154+
both this and the bound above, and would let one worker's stamps be used instead of
155+
a mix. That is compute-side work.
156+
157+
Multi-export dataflows would need one more step. All exports of a dataflow share a
158+
suspension token, so the dataflow starts only once every export is scheduled, while
159+
the stamp is per export. Every compute dataflow has exactly one export today and
160+
`sequential_hydration.rs` asserts it, so the two coincide.
161+
162+
## What is and is not recorded
163+
164+
Collection samples current state. It is not an event log, and the log it reads
165+
retracts an object's row when the export goes away. An episode is recorded only if
166+
its row is still live when its replica's turn comes around, so an object dropped
167+
before then, or a replica process that restarts before then, leaves no trace. An
168+
object that never reports a completion time, such as a constant materialized view
169+
whose frontier jumps straight to empty, is never recorded.
170+
`test/testdrive/hydration-status.td` asserts that absence by name, so the limit is
171+
pinned rather than merely tolerated.
172+
173+
Making these durable requires compute to emit hydration transitions into an
174+
append-only collection that survives until an observer acknowledges them. Until
175+
then, best effort is the honest description of a sampler.
176+
177+
## Retention
178+
179+
`hydration_history_retention_period` defaults to 30 days. Retention is another OCC
180+
mutation: it subscribes to rows older than the cutoff and writes their retractions
181+
at the observed frontier. Collection applies the same cutoff, so a still-live log
182+
row cannot resurrect an episode retention just retracted.
183+
184+
Retention deletes a bounded batch per sweep and converges over as many sweeps as it
185+
takes. The bound is not a nicety. The OCC path refuses a selection larger than
186+
`max_result_size` before submitting any write, so one unbounded delete over a large
187+
backlog would fail identically forever and never shrink the table. The bound has to
188+
sit inside a derived table, because a top-level `LIMIT` lands in the plan's
189+
`RowSetFinishing`, which the OCC path deliberately discards, and the delete would
190+
be silently unbounded again.
191+
192+
Retention runs on the catalog server, so it keeps working when there are no user
193+
replicas at all, and it runs even when that sweep's collection failed. A
194+
crash-looping replica must not be able to stop the table from shrinking. The
195+
dependency does not run the other way: a catalog server without a replica skips
196+
retention and leaves collection running.
197+
198+
Disabling collection also suspends retention. The alternative is an always-on
199+
subscribe in the default configuration, where the table is empty and there is
200+
nothing to retain. Rows already collected are therefore kept while collection is
201+
off.
202+
203+
## Durability is best effort
204+
205+
Builtin tables are reset at bootstrap and re-shard on a forced schema migration.
206+
The history table is exempt from both, because a sampled history cannot be rebuilt
207+
from anything else once it is gone. Those two exemptions are the whole promise.
208+
209+
We do not commit to carrying the contents across every future version. A schema
210+
change to a builtin table allocates a fresh shard, so changing these columns clears
211+
the history. That is an acceptable trade: the value here is the distribution the
212+
table accumulates, and starting it over costs one retention period, while freezing
213+
the schema to protect it costs every later improvement.
214+
215+
So we try not to break it and we do not promise not to. The exemption exists to
216+
stop an incidental migration from wiping the table as a side effect of unrelated
217+
work, not to make it untouchable. Giving it up is deliberate: an assert in
218+
`plan_migration` fires if a migration step ever names this table, and the comment
219+
there says to remove the assert and the exemption together and to note in the
220+
release notes that the history restarts. The user-facing documentation states the
221+
same limit, that a schema change in a future release may clear the contents.
222+
223+
## Scheduling and isolation
224+
225+
`hydration_history_collection_interval` sets the cadence and disables collection at
226+
zero. Sweeps never overlap, which bounds compute load and keeps the collector from
227+
contending with itself in the serialized timestamped-write path.
228+
229+
Fires align to interval boundaries, and each sleep is capped, so that lowering a
230+
long interval at runtime takes effect within the cap rather than after the old
231+
interval elapses. Tests depend on that. A disabled collector polls far more
232+
coarsely, since that is the cadence of every environment in the default
233+
configuration.
234+
235+
Two isolation decisions are worth stating outright:
236+
237+
**Background mutations take no OCC write permit.** A session's wait for a permit is
238+
bounded by its statement timeout. A sweep has no statement timeout, and its
239+
subscribe must first hydrate a dataflow on a user replica, so holding a permit
240+
would let a background sampler stall user DML for as long as that takes. The sweep
241+
is single-flight, so skipping the permit adds at most one concurrent
242+
read-then-write.
243+
244+
**The sweep is aborted when the coordinator is dropped.** Unlike a session it holds
245+
no `Client`, so nothing otherwise stops it from outliving the coordinator, and the
246+
runtime teardown that follows drops the timestamp oracle's worker task. A sweep
247+
still running then reads a timestamp from a dead oracle and panics. The coordinator
248+
owns the task handle so that dropping it cancels the sweep first.
249+
250+
Each mutation has its own deliberately generous timeout. Even a mutation with
251+
nothing to write waits for its read to linearize, which can take a full
252+
`default_timestamp_interval`, a parameter with no upper bound, so a tighter bound
253+
would let a large timestamp interval starve retention permanently.
254+
255+
Replica drop, cluster drop, dependency replacement, replica failure, and timeout all
256+
skip the attempt, and a later sweep recomputes from current state. Read-only
257+
generations do nothing. Replicas with introspection disabled are skipped, since
258+
their log arrangements exist but are never populated, so a subscribe would read a
259+
sealed, empty collection on every sweep forever.
260+
261+
## Known limitation: frontier skew
262+
263+
The write timestamp is the subscribe's observed frontier rather than a fresh oracle
264+
timestamp, for the reason given above. That frontier is the minimum over the
265+
subscribe's inputs, and one input is a replica-local introspection log whose
266+
frontier the *replica* advances from its own clock, rounded up to its introspection
267+
interval.
268+
269+
So a replica whose clock trails environmentd by more than one introspection
270+
interval produces a frontier that never gets ahead of the timestamp oracle. Every
271+
attempt loses its timestamp race and the step exhausts its retries. The consequence
272+
is bounded. Nothing wrong is recorded, that replica records nothing at all, and the
273+
sweep stretches while it retries, delaying others in the rotation. It resolves
274+
itself when the clocks converge.
275+
276+
Accepted for now: skew between processes is normally milliseconds while the
277+
tolerance is a whole introspection interval, and the failure mode is silent retry
278+
rather than wrong data. Because the symptom is otherwise hard to attribute,
279+
exhausting retries logs that the replica's introspection frontier may be trailing.
280+
Removing the limitation means deriving the write timestamp from the target table's
281+
own frontier, which needs a correctness argument for applying replica-derived diffs
282+
at a timestamp the replica never observed.
283+
284+
## Notes on the catalog plumbing
285+
286+
The `test/sqllogictest/autogenerated/*.slt` goldens are generated from the
287+
user-facing docs markdown, but the sqllogictest run compares them against the live
288+
catalog. Editing a column comment in the markdown without editing it in the Rust
289+
builtin definition produces a golden that passes the docs lint locally and fails in
290+
CI.
291+
292+
## Rollout
293+
294+
The interval is zero, and so disabled, in production, and runtime configuration can
295+
enable it without restarting environmentd. The mzcompose configuration enables it at
296+
60 seconds so hydration, restart, retention, and catalog tests exercise the path.
297+
298+
It stays off in the sqllogictest runner defaults, against the usual preference for
299+
enabling new paths in tests. The collector installs subscribes and writes a builtin
300+
table, while those runs assert on catalog contents and plans, so enabling it risks
301+
churn and timing flakiness in files that have nothing to do with hydration.
302+
303+
## Future Work
304+
305+
- Record installation and start before completion, then finalize canceled and
306+
failed episodes by joining replica lifecycle events.
307+
- Define a replica episode state machine from the object set present at the
308+
hydrated-to-hydrating transition.
309+
- Publish resettable per-process high-water values for RAM, swap, and scratch disk,
310+
and define replica aggregation without pretending sampled maxima are simultaneous
311+
peaks.
312+
- Give storage objects equivalent lifecycle timestamps.
313+
- Build replica history and progress views on those signals.

0 commit comments

Comments
 (0)