From 1a6f46f3fa2584d96c9bb2173e4013796cb27904 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 21 Aug 2026 16:37:12 +0000 Subject: [PATCH] compute: add a lifecycle event log for compute exports 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 Claude-Session: https://claude.ai/code/session_018ZVCMBSLdxzGus78ZKWhZz --- .../20260817_compute_hydration_timestamps.md | 261 ++++++++++++++++-- .../system-catalog/mz_introspection.md | 1 + .../catalog/open/builtin_schema_migration.rs | 22 ++ src/catalog/src/builtin.rs | 16 ++ src/catalog/src/builtin/mz_introspection.rs | 35 +++ src/catalog/src/durable/transaction.rs | 1 + src/compute-client/src/logging.rs | 14 + src/compute/src/compute_state.rs | 146 +++++++++- src/compute/src/logging/compute.rs | 217 ++++++++++++++- src/compute/src/sink/materialized_view.rs | 14 +- src/compute/src/sink/materialized_view_v2.rs | 6 +- src/pgrepr-consts/src/oid.rs | 1 + .../autogenerated/mz_introspection.slt | 1 + test/sqllogictest/catalog_server_explain.slt | 14 +- test/sqllogictest/cluster.slt | 12 +- test/sqllogictest/cockroach/srfs.slt | 36 +++ test/sqllogictest/distinct_arrangements.slt | 1 + .../information_schema_tables.slt | 4 + test/sqllogictest/introspection/relations.slt | 6 +- .../mz_catalog_server_index_accounting.slt | 169 ++++++------ test/sqllogictest/oid.slt | 1 + test/sqllogictest/pg_catalog_user.slt | 2 +- test/sqllogictest/regclass.slt | 12 +- test/sqllogictest/regtype.slt | 4 +- test/testdrive/catalog.td | 3 +- test/testdrive/compute-lifecycle-events.td | 168 +++++++++++ test/testdrive/indexes.td | 1 + test/workload-replay/objects.txt | 7 + .../system_catalog_identifiers.txt | 7 + 29 files changed, 1039 insertions(+), 143 deletions(-) create mode 100644 test/testdrive/compute-lifecycle-events.td diff --git a/doc/developer/design/20260817_compute_hydration_timestamps.md b/doc/developer/design/20260817_compute_hydration_timestamps.md index 0ac81a5990ac5..5a0f271f406be 100644 --- a/doc/developer/design/20260817_compute_hydration_timestamps.md +++ b/doc/developer/design/20260817_compute_hydration_timestamps.md @@ -103,12 +103,26 @@ them. | --- | --- | --- | | `installed_at` | the `Export` event, from `CreateDataflow` | the dataflow exists on this worker, suspended, so this is also the start of queueing | | `started_at` | the dataflow is unsuspended | hydration is actually running | -| `hydrated_at` | the output frontier passes the as-of | hydration is complete | +| `hydrated_at` | the reported output frontier passes the as-of | the output is readable, which for a collection that writes means durable | `hydrated_at - started_at` is hydration time as users mean it, and `started_at - installed_at` is the queueing interval. Today's `time_ns` conflates the two. +The lifecycle is wider than these three stages, and timestamp columns cannot carry +all of it. Two things get in the way. 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. And a timestamp column 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. + +So the lifecycle proper is recorded as an append-only event log, described under +"The lifecycle event log", and `mz_compute_hydration_times_per_worker` keeps +exactly the shape and meaning above. The two are complementary: the timestamps are +the compact per-worker summary a rollup aggregates, and the log is where causes and +the write stages live. + Two choices shape everything else, each argued in its own section below. The timestamps are stamped by the replica rather than by the compute controller, for the reasons in "Why the replica and not the compute controller". And `time_ns` is @@ -163,6 +177,142 @@ compute logging and has not been observed to be severe, but it is a real risk an this design is the first to invite direct comparison of absolute times, so it is acknowledged rather than designed around. +### The lifecycle event log + +One append-only log relation, per replica, in memory: + +``` +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 +``` + +| event | grain | `reason` | +| --- | --- | --- | +| `installed` | per worker | none | +| `started` | per worker | none | +| `hydrated` | per worker | none | +| `write_blocked` | per object | `read_only` | +| `write_unblocked` | per object | none | +| `written` | per object | none | + +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. Subscribes and +`COPY TO` stop 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. + +**Grain.** `worker_id` is 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, elected as `hashed(sink_id) % peers`, so they appear once +per object rather than once per worker, and the row records which worker was +elected for free. Nothing else may read that shared frontier as a measure of +writing: `mint` clears it on every non-elected worker, where it is then the empty +antichain and would report having written everything immediately. The election has +one definition, `sink::materialized_view::frontier_owner`, called both by `mint` +and by the code that records ownership. + +**Which frontier each stage reads.** `hydrated` reads the dataflow's own progress +frontier, the compute probe, not the reported output frontier. The output frontier +is the meet of write and compute frontier, which makes it a measure of durability +rather than of computation, and for a sink-backed collection it is not even uniform +across workers, for the reason just given. A collection with no compute probe +produces its output *by* writing it, an index into its own trace, so there the +write frontier is the progress and `hydrated` coincides with durability. + +`written` reads the sink's write frontier passing the as-of, held back until +`hydrated` has been reported and until writes are permitted. Without the first +clamp the two can invert, for the reason under "Refresh schedules do not block +writing". + +The second clamp is there because the frontier is the output shard's upper, which +is a property of the shard rather than of this replica. The as-of is bounded to one +step below that upper for a non-empty storage export, in +`as_of_selection::apply_downstream_storage_constraints`, so for a shard that +already holds data the predicate is true from the moment the dataflow is installed. +A replica that may not write cannot be the one that advanced it, so reporting the +stage there would attribute another writer's progress to this replica and put +`written` ahead of `write_unblocked`. What `written` promises is therefore that the +output is durable at the as-of and that this replica was permitted to write, not +that this replica performed the write. On a restarted or scaled-out replica the +output was already durable, and `written` lands with `hydrated`. + +**`write_blocked` is logged on entry, not on exit.** Carrying the cause on +`write_unblocked` reads more naturally, but then the cause is only observable once +the block has ended. If it never ends, which is the state an operator is debugging, +there is no row at all. Logging entry makes "which objects are hydrated but not +writing, and why" a query over present rows rather than an inference from absence. + +Entry means entry into a state where the block matters, which is after hydration. +Before it, the sink has produced nothing and read-only mode is holding nothing +back. Every collection starts read-only and is released by the controller, so +reporting a block from installation would put a `write_blocked` and a +`write_unblocked` on essentially every materialized view, both before `hydrated`, +making `write_unblocked - hydrated` negative rather than zero. Gating on hydration +means the pair appears only when something really was held back, and the intervals +in the list above are all non-negative by construction. + +**`write_unblocked`, not `write_started`.** `mint` produces a batch description as +soon as the desired frontier advances past the persist frontier, and the persist +frontier is initialized to the as-of, so for a plain read-write materialized view +the first write is minted at hydration and a separate "write started" stamp would +carry no information. What does vary is when writing became *permitted*. Naming it +that way gives each interval exactly one cause: `started - installed` is queueing, +`hydrated - started` is compute, and `write_unblocked - hydrated` is blocked time. +In the common case the blocked pair is absent rather than zero: the controller +allows writes in the same turn it ships the dataflow, so a collection is normally +already permitted to write by the time it hydrates, and neither event is logged. + +**`reason` is typed, `details` is not.** This follows `mz_source_statuses` and +`mz_sink_statuses`, which pair a typed status with a nullable `details jsonb` +documented by example rather than by schema. What people filter and group on stays +typed, and only the look-at-one-row detail goes in the json. What `details` carries +is the dataflow's as-of, which every stage is defined relative to: without it +`hydrated - started` cannot distinguish a genuinely fast hydration from one whose +as-of was already recent, and a replacement materialized view with a far behind +as-of is a completely different amount of work at the same duration. That does not +earn a column and is worth having in the row. Invariant tests must assert on +`event`, `reason` and `occurred_at` and never on `details`, or the first test that +pins a field removes the extensibility it exists for. + +**Bounds.** At most six rows per object, times workers for the first three events, +all retracted when the object is dropped. This is in-memory introspection, so +there is no durable growth to reason about. + +**Only `read_only` is attributed.** It is the one cause of a write block that +compute can observe. Two further attributions would be useful and are not +available, so they are follow-up work rather than part of this design. +Distinguishing a `started` that waited on the hydration limiter from one that +waited on its inputs needs `SequentialHydration` to report which, since both appear +to the replica as `Schedule` arriving late. And distinguishing a dataflow installed +by a fresh `CreateDataflow` from one retained across reconciliation is not +observable here at all, because a retained dataflow emits no new `installed` event. + +### Refresh schedules do not block writing + +`apply_refresh` rounds a `REFRESH` materialized view's frontier *up* to the next +refresh time, and it does so off its input frontier, before the dataflow has +computed anything. The sink therefore sees a desired frontier ahead of the as-of +immediately, mints a description for the pre-refresh window, and appends an empty +batch, advancing the shard's upper. A refresh schedule brings writing forward +rather than holding it back. + +`test/testdrive/materialized-view-refresh-options.td` shows this from the outside: +a materialized view whose first refresh is far in the future reports +`mz_hydration_statuses.hydrated = true`, and that flag is `time_ns IS NOT NULL`, +which requires the write frontier to have passed the as-of. + +Two consequences. There is no `refresh` cause for `write_blocked` to report, +because there is no such state. And the shard's upper can pass the as-of while the +dataflow is still hydrating, which is why `written` is clamped to `hydrated`. + +What a refresh schedule does still distort is any rollup reading +`mz_compute_hydration_statuses.hydration_time` as hydration work, since for a +refresh materialized view that interval can include waiting on the schedule. That +is a property of the retained `time_ns` column, not of the log. + ### A new hydration start event There is no event for hydration start today. Add @@ -274,10 +424,14 @@ change nor the rename would have touched that relation. **`time_ns` is kept rather than replaced.** It is the reason the existing columns keep their exact values: retained rather than derived, so nothing is recomputed, no precision is lost, and no cross-worker arithmetic is introduced. -Deriving `time_ns` as `hydrated_at - installed_at` would have moved it to +It could not be derived from the timestamps in any case. `time_ns` and +`hydrated_at` fire on the same crossing, but deriving the duration would move it to microsecond precision, since `timestamptz` caps there, where today it is true -nanoseconds. Deriving it after aggregation would additionally have absorbed -cross-worker install skew and the per-worker anchor skew described above. So +nanoseconds. It would also change what the interval is measured from: `time_ns` +runs off a single `Instant` taken when the export state is created, where +`hydrated_at - installed_at` is a difference of two rounded event times. Deriving +it after aggregation would additionally have absorbed cross-worker install skew and +the per-worker anchor skew described above. So `time_ns` remains the authoritative per-worker duration, measured from a single `Instant` inside one worker, and the timestamps carry episode identity, which requires absolute times a duration cannot provide. Two columns with two documented @@ -312,7 +466,7 @@ in one controller turn and one replica turn. | 6 | replica | inserts the suspension token and renders the dataflow, whose operators park on the `StartSignal` | | | 7 | replica | `handle_schedule` drops the token and the operators start | **`started_at`** | | 8 | replica | the dataflow reads its inputs from the as-of forward and builds arrangements. Nothing is stamped here, this interval is the hydration | | -| 9 | replica | the output frontier passes the as-of and `set_reported_output_frontier` calls `set_hydrated` | **`hydrated_at`** | +| 9 | replica | the reported output frontier passes the as-of and `set_reported_output_frontier` calls `set_hydrated`. Separately, `observe_hydration` sees the dataflow's own progress frontier pass the as-of and logs the `hydrated` stage | **`hydrated_at`**, and the `hydrated` event | | 10 | replica | the demux writes the retract and insert pair, so the per-worker relation carries all three | | | 11 | controller | separately, a `Frontiers` response arrives and `update_output_frontier` flips the controller's own hydration view, which is what the 0dt caught-up check and the autoscaling signal read. One round trip later, and it stamps nothing | | @@ -391,21 +545,16 @@ restarting. Consumers must gate on introspection freshness, as `mz_object_arrangement_size_history` already does via `fresh_introspection_replicas`. -**`REFRESH` materialized views report a refresh interval, not hydration work.** -The reported output frontier is the meet of write and compute frontier, and a -REFRESH MV's write frontier sits at the as-of until the first refresh lands, so -hydration is not considered complete until then. For `REFRESH EVERY '1 day'`, -`hydrated_at - started_at` can be most of a day, nearly all of it idle. The -per-object stamps are still internally consistent, so this is not a defect in the -relation, but any rollup must exclude these objects or it will never close an -episode. The controller already receives `refresh_schedule` in `add_collection`, -so it can mark them. - -**Read-only mode changes what the output frontier means.** In read-only mode the -write frontier is deliberately excluded from the reported output frontier, because -a read-only dataflow cannot push it forward. So `hydrated_at` during a 0dt -read-only window reflects compute progress only, which is the intended reading but -differs from the steady-state one. +**`REFRESH` materialized views hydrate on their computation.** The compute probe +is attached before the `apply_refresh` operator, deliberately, with the comment in +`src/compute/src/sink/materialized_view.rs` explaining that rounding frontiers up +"makes it impossible to accurately track the progress of the computation". So the +log's `hydrated` stage reads the pre-rounding frontier. `hydrated_at` agrees, even +though it reads the meet: a refresh schedule pushes the write frontier ahead of the +as-of, so the meet is bounded by the compute frontier and crosses when the +computation does. Both report when the computation caught up rather than anything +derived from the schedule. What the schedule does affect is writing, and it +advances it rather than delaying it. See "Refresh schedules do not block writing". ### Why the replica and not the compute controller @@ -435,6 +584,8 @@ for upgrades. ### Implementation touch points +The timestamps: + - `src/compute/src/logging/compute.rs`: `ComputeEvent::HydrationStart`, the three `ExportState` fields, the packer, `handle_export`, `handle_export_dropped`, `handle_hydration` including the `started_at` backfill, and a new @@ -448,12 +599,50 @@ for upgrades. the variant itself is unchanged. - `src/catalog/src/builtin/mz_introspection.rs`: the appended columns on the existing builtin log. No rename, so no new OID and no `BUILTINS_STATIC` entry. -- Goldens that hardcode this relation's identity, columns, OIDs or indexes: - `test/sqllogictest/oid.slt`, `information_schema_tables.slt`, - `mz_catalog_server_index_accounting.slt`, `cluster.slt`, - `catalog_server_explain.slt`, `test/cluster/mzcompose.py`, and the autogenerated - `test/sqllogictest/autogenerated/mz_introspection.slt`. -- Docs: the `mz_introspection` system catalog reference page. + +The lifecycle log: + +- `src/compute-client/src/logging.rs`: a `ComputeLog::LifecycleEvent` variant and + its `RelationDesc`, unkeyed, so `index_by` arranges by the whole row. Declaring + `(export_id, worker_id, event)` as a key would be true today but is a uniqueness + claim the optimizer would act on, and a false key is a correctness hazard rather + than a missed optimization. +- `src/catalog/src/durable/transaction.rs`: a new log id. Existing ids must not be + renumbered. Doing so panics on restart with a negative capability on + `IntrospectionSourceIndex`. +- `src/catalog/src/builtin/mz_introspection.rs` and `src/catalog/src/builtin.rs`: a + `BuiltinLog` with a fresh OID and an ontology entry, plus its `BUILTINS_STATIC` + registration. +- `src/compute/src/logging/compute.rs`: the `Lifecycle` event and the + `LifecycleStage` vocabulary, an as-of field on `Export`, a demux output and + packer including the `jsonb` column, and the emitted rows kept on `ExportState` + so that they can be retracted verbatim when the export is dropped. +- `src/compute/src/compute_state.rs`: the stage bookkeeping on `CollectionState` + and the observation of both frontiers in `report_frontiers`. +- `src/compute/src/sink/materialized_view.rs` and `materialized_view_v2.rs`: the + shared `frontier_owner` election, and recording on the collection whether this + worker owns the sink frontier. + +- `src/adapter/src/catalog/open/builtin_schema_migration.rs`: `Replacement` steps for + `mz_catalog.mz_indexes` and `mz_catalog.mz_sources` at the workspace's current dev + version. Adding a builtin log moves two generated materialized views. `make_mz_indexes` + inlines one `VALUES` row per log, naming the log and its `index_by` columns, and + `make_mz_sources` inlines one per log alongside the builtin sources. Either fingerprint + moving without a step reaches `update_fingerprints` with a mismatch for a builtin that is + neither migrated nor ephemeral, which panics and blocks catalog open on upgrade. + +Goldens that hardcode a log relation's identity, columns, OIDs or indexes: +`test/sqllogictest/oid.slt`, `information_schema_tables.slt`, +`mz_catalog_server_index_accounting.slt`, `cluster.slt`, +`cockroach/srfs.slt`, the autogenerated +`test/sqllogictest/autogenerated/mz_introspection.slt`, +`test/testdrive/indexes.td`, `test/testdrive/catalog.td`, and +`test/workload-replay/system_catalog_identifiers.txt` and `objects.txt`. Docs: the +`mz_introspection` system catalog reference page. + +`catalog_server_explain.slt` and `test/cluster/mzcompose.py` need no change. The +former's query filters `o.id NOT LIKE 'si%'`, which excludes per-replica +introspection log indexes, and the latter queries named relations. Not touched, and deliberately so: the introspection subscribe, `mz_internal.mz_compute_hydration_times`, @@ -621,8 +810,24 @@ named. visibility limit. - **Log collections** get a start event rather than a filter. - **The per-replica relation is follow-up work,** not part of this design. +- **The lifecycle is an event log, not more timestamp columns.** The stages do not + share a grain and a timestamp cannot carry a cause. See "The lifecycle event + log". +- **`mz_compute_hydration_times_per_worker` keeps its meaning.** `hydrated_at` + reads the output frontier, as it always has, so nothing built on it changes + value. The log carries the dataflow reading under `hydrated`. +- **Refresh schedules advance writing rather than blocking it,** established + against the refresh tests. So `write_blocked` has no `refresh` cause, and + `written` is clamped to `hydrated` to keep the stages ordered. See "Refresh + schedules do not block writing". +- **`worker_id` is not nullable.** A NULL would make the per-object grain visible + in the row, at the cost of introducing the only NULL `worker_id` in the logging + framework. The grain is documented instead, and the column records which worker + was elected. ## Open questions -None outstanding for this design. The open questions all belong to the per-replica -rollup and are enumerated under "Follow-up work". +None outstanding for this design. Two attributions the `reason` vocabulary would +benefit from are not observable today and are enumerated under "The lifecycle event +log". The rest of the open questions belong to the per-replica rollup and are +enumerated under "Follow-up work". diff --git a/doc/user/content/reference/system-catalog/mz_introspection.md b/doc/user/content/reference/system-catalog/mz_introspection.md index f238ce5e8240b..880be9b10b90f 100644 --- a/doc/user/content/reference/system-catalog/mz_introspection.md +++ b/doc/user/content/reference/system-catalog/mz_introspection.md @@ -464,6 +464,7 @@ The `mz_scheduling_parks_histogram` view describes a histogram of [dataflow] wor [query hints]: /sql/select/#query-hints + diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index 8af544edca8b5..a505103b920b3 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -416,6 +416,28 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_CATALOG_SCHEMA, "mz_views", ), + // Required because we added the `mz_compute_lifecycle_events_per_worker` builtin log. + // make_mz_indexes inlines one VALUES row per builtin log, naming the log and its + // `index_by` columns, so adding or removing a log changes the SQL fingerprint of + // `mz_indexes` just as adding a builtin index does. See the NOTE above: this version + // must stay at the workspace's current dev version until the change ships. + MigrationStep::replacement( + "26.40.0-dev.0", + CatalogItemType::MaterializedView, + MZ_CATALOG_SCHEMA, + "mz_indexes", + ), + // Adding a builtin log moves two generated materialized views, not one: + // make_mz_sources inlines a VALUES row per builtin log alongside the builtin sources, + // so `mz_sources` needs the same treatment. Without it, an upgrade from a released + // version reaches `update_fingerprints` with a mismatch for a builtin that is neither + // migrated nor ephemeral, which panics and blocks catalog open. + MigrationStep::replacement( + "26.40.0-dev.0", + CatalogItemType::MaterializedView, + MZ_CATALOG_SCHEMA, + "mz_sources", + ), ] }); diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs index e7eeb0ccd017f..395cb7aefbf09 100644 --- a/src/catalog/src/builtin.rs +++ b/src/catalog/src/builtin.rs @@ -1106,6 +1106,7 @@ pub static BUILTINS_STATIC: LazyLock>> = LazyLock::ne Builtin::Log(&MZ_COMPUTE_IMPORT_FRONTIERS_PER_WORKER), Builtin::Log(&MZ_COMPUTE_ERROR_COUNTS_RAW), Builtin::Log(&MZ_COMPUTE_HYDRATION_TIMES_PER_WORKER), + Builtin::Log(&MZ_COMPUTE_LIFECYCLE_EVENTS_PER_WORKER), Builtin::Log(&MZ_COMPUTE_OPERATOR_HYDRATION_STATUSES_PER_WORKER), Builtin::MaterializedView(&MZ_KAFKA_SINKS), Builtin::MaterializedView(&MZ_KAFKA_CONNECTIONS), @@ -2220,6 +2221,21 @@ mod tests { Fingerprint::fingerprint(&&mv_extra), "mz_sources fingerprint must change when a builtin source is added" ); + + // Adding an extra log must also change the fingerprint, because the log set is inlined + // alongside the source set. Without this case, adding a builtin log moves the + // `mz_sources` fingerprint with nothing on the PR path to announce that it needs a + // migration step, and catalog open panics on the upgrade. + let extra_log = logs[0]; + let mv_extra_log = builtin::make_mz_sources( + sources.iter().copied(), + logs.iter().copied().chain(std::iter::once(extra_log)), + ); + assert_ne!( + fp_base, + Fingerprint::fingerprint(&&mv_extra_log), + "mz_sources fingerprint must change when a builtin log is added" + ); } /// Verifies that the `mz_indexes` materialized view fingerprint changes diff --git a/src/catalog/src/builtin/mz_introspection.rs b/src/catalog/src/builtin/mz_introspection.rs index d5ad93b478809..f2028c6639d56 100644 --- a/src/catalog/src/builtin/mz_introspection.rs +++ b/src/catalog/src/builtin/mz_introspection.rs @@ -342,6 +342,41 @@ pub static MZ_COMPUTE_HYDRATION_TIMES_PER_WORKER: LazyLock = }), }); +pub static MZ_COMPUTE_LIFECYCLE_EVENTS_PER_WORKER: LazyLock = + LazyLock::new(|| BuiltinLog { + name: "mz_compute_lifecycle_events_per_worker", + schema: MZ_INTROSPECTION_SCHEMA, + oid: oid::LOG_MZ_COMPUTE_LIFECYCLE_EVENTS_PER_WORKER_OID, + variant: LogVariant::Compute(ComputeLog::LifecycleEvent), + access: vec![PUBLIC_SELECT], + ontology: Some(Ontology { + entity_name: "compute_lifecycle_event_per_worker", + description: "Lifecycle events of each compute export, as observed by the worker \ + that logged them. Every event carries the wallclock instant it \ + occurred at, so durations between stages are differences of \ + `occurred_at`. The `installed`, `started` and `hydrated` events are \ + logged by every worker, since each worker hydrates its own fragment \ + of the dataflow. The write events are logged only by the worker that \ + maintains the sink frontier, so they appear once per export rather \ + than once per worker.", + links: &const { + [OntologyLink { + name: "lifecycle_event_of", + target: "compute_export_per_worker", + properties: LinkProperties::MapsTo { + source_column: "export_id", + target_column: "export_id", + via: None, + from_type: Some(SemanticType::GlobalId), + to_type: Some(SemanticType::GlobalId), + note: None, + }, + }] + }, + column_semantic_types: &[("export_id", SemanticType::GlobalId)], + }), + }); + pub static MZ_COMPUTE_OPERATOR_HYDRATION_STATUSES_PER_WORKER: LazyLock = LazyLock::new(|| BuiltinLog { name: "mz_compute_operator_hydration_statuses_per_worker", diff --git a/src/catalog/src/durable/transaction.rs b/src/catalog/src/durable/transaction.rs index c27c43fadcc93..33bbe5775791a 100644 --- a/src/catalog/src/durable/transaction.rs +++ b/src/catalog/src/durable/transaction.rs @@ -1021,6 +1021,7 @@ impl<'a> Transaction<'a> { LogVariant::Compute(ComputeLog::DataflowGlobal) => 31, LogVariant::Compute(ComputeLog::OperatorHydrationStatus) => 32, LogVariant::Compute(ComputeLog::PrometheusMetrics) => 33, + LogVariant::Compute(ComputeLog::LifecycleEvent) => 34, }; let mut id: u64 = u64::from(cluster_variant) << 56; diff --git a/src/compute-client/src/logging.rs b/src/compute-client/src/logging.rs index d3b788af63efe..9a31907f262c5 100644 --- a/src/compute-client/src/logging.rs +++ b/src/compute-client/src/logging.rs @@ -176,6 +176,8 @@ pub enum ComputeLog { ErrorCount, /// Hydration times of exported collections. HydrationTime, + /// Lifecycle events of exported collections. + LifecycleEvent, /// Hydration status of dataflow operators. OperatorHydrationStatus, /// Mappings from `GlobalId`/`LirId`` pairs to dataflow addresses. @@ -370,6 +372,18 @@ impl LogVariant { .with_key(vec![0, 1]) .finish(), + LogVariant::Compute(ComputeLog::LifecycleEvent) => RelationDesc::builder() + .with_column("export_id", SqlScalarType::String.nullable(false)) + .with_column("worker_id", SqlScalarType::UInt64.nullable(false)) + .with_column("event", SqlScalarType::String.nullable(false)) + .with_column( + "occurred_at", + SqlScalarType::TimestampTz { precision: None }.nullable(false), + ) + .with_column("reason", SqlScalarType::String.nullable(true)) + .with_column("details", SqlScalarType::Jsonb.nullable(true)) + .finish(), + LogVariant::Compute(ComputeLog::OperatorHydrationStatus) => RelationDesc::builder() .with_column("export_id", SqlScalarType::String.nullable(false)) .with_column("lir_id", SqlScalarType::UInt64.nullable(false)) diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index 3a2f7b42e2697..a9363068ea158 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -70,7 +70,7 @@ use uuid::Uuid; use crate::arrangement::manager::{TraceBundle, TraceManager}; use crate::logging; -use crate::logging::compute::{CollectionLogging, ComputeEvent, PeekEvent}; +use crate::logging::compute::{CollectionLogging, ComputeEvent, LifecycleStage, PeekEvent}; use crate::logging::initialize::LoggingTraces; use crate::metrics::{CollectionMetrics, WorkerMetrics}; use crate::render::{LinearJoinSpec, StartSignal}; @@ -694,6 +694,7 @@ impl<'a> ActiveComputeState<'a> { object_id, logger, *dataflow_index, + as_of.as_option().copied(), dataflow.import_ids(), ); if starts_immediately { @@ -884,12 +885,17 @@ impl<'a> ActiveComputeState<'a> { let mut collection = CollectionState::new( Rc::clone(&dataflow_index), is_subscribe_or_copy, - as_of, + as_of.clone(), metrics, ); - let logging = - CollectionLogging::new(id, logger.clone(), *dataflow_index, std::iter::empty()); + let logging = CollectionLogging::new( + id, + logger.clone(), + *dataflow_index, + as_of.as_option().copied(), + std::iter::empty(), + ); // Log collections are never suspended and the controller marks them scheduled // implicitly, so no `Schedule` command ever arrives for them. Record their hydration // start here, or they would sit permanently in the illegal state of being hydrated @@ -921,6 +927,8 @@ impl<'a> ActiveComputeState<'a> { // Maintain a single allocation for `new_frontier` to avoid allocating on every iteration. let mut new_frontier = Antichain::new(); + // Same, for the frontier that measures dataflow progress. + let mut hydration_frontier = Antichain::new(); for (&id, collection) in self.compute_state.collections.iter_mut() { // The compute protocol does not allow `Frontiers` responses for subscribe and copy-to @@ -950,6 +958,36 @@ impl<'a> ActiveComputeState<'a> { .allows_reporting(&new_frontier) .then(|| new_frontier.clone()); + // Collect the frontier that measures the dataflow's own progress, which is what + // hydration is about. + // + // This is deliberately not the output frontier collected below. That folds in the + // write frontier, which makes it a measure of durability rather than of dataflow + // progress, and for a collection that sinks to persist it is not even uniform across + // workers: the sink's `mint` operator maintains the shared sink frontier on one + // elected worker and clears it on all the others, so the same dataflow would report + // hydration at two different times depending on which worker's log you read. + // + // A collection with a compute frontier produces its output before writing it, so that + // frontier is its progress. A collection without one produces its output *by* writing + // it, an index into its own trace, so there the write frontier is the progress and + // hydration coincides with durability. + hydration_frontier.clear(); + match &collection.compute_probe { + Some(probe) => { + probe.with_frontier(|frontier| { + hydration_frontier.extend(frontier.iter().copied()) + }); + } + None => hydration_frontier.clone_from(&new_frontier), + } + + // Evaluate the lifecycle predicates here, while both frontiers are still in hand. + // `new_frontier` is folded into the output frontier below, which loses the write + // frontier this one is about. + let hydrated = PartialOrder::less_than(&collection.as_of, &hydration_frontier); + let written = PartialOrder::less_than(&collection.as_of, &new_frontier); + // Collect the output frontier and check for progress. // // By default, the output frontier equals the write frontier (which is still stored in @@ -996,6 +1034,9 @@ impl<'a> ActiveComputeState<'a> { .set_reported_output_frontier(ReportedFrontier::Reported(frontier.clone())); } + collection.observe_hydration(hydrated); + collection.observe_writes(written); + let response = FrontiersResponse { write_frontier: new_write_frontier, input_frontier: new_input_frontier, @@ -1209,6 +1250,12 @@ impl<'a> ActiveComputeState<'a> { .set_reported_write_frontier(ReportedFrontier::Reported(new_frontier.clone())); collection .set_reported_input_frontier(ReportedFrontier::Reported(new_frontier.clone())); + // Only a batch upper measures progress. `DroppedAt` reports the empty + // antichain, which is the maximum of the order, so a subscribe cancelled while + // still hydrating would otherwise read as hydrated at the moment it is dropped. + let hydrated = matches!(response, SubscribeResponse::Batch(_)) + && PartialOrder::less_than(&collection.as_of, &new_frontier); + collection.observe_hydration(hydrated); collection.set_reported_output_frontier(ReportedFrontier::Reported(new_frontier)); } else { // Presumably tracking state for this subscribe was already dropped by @@ -1998,6 +2045,21 @@ pub struct CollectionState { logging: Option, /// Metrics tracked for this collection. metrics: CollectionMetrics, + /// Whether this worker maintains the authoritative write frontier of this collection's sink. + /// + /// A persist sink elects one worker to track the output shard's upper and clears the shared + /// frontier on all the others, so only the elected worker's copy carries write progress. The + /// write lifecycle stages are logged by that worker alone, which also makes them a single + /// observation per export rather than one per worker. False for collections whose output + /// frontier is not a persist upper at all, such as indexes and metric sinks. + pub owns_sink_frontier: bool, + /// Which lifecycle stages have been logged for this collection. + /// + /// Stages are only ever added, never removed. Reconciliation resets the reported frontiers of + /// a retained dataflow, so without this the collection would look unhydrated again and re-log + /// a stage it already reported. The lifecycle relation is append-only, so a repeat would show + /// up as a duplicate row rather than being dropped. + logged_stages: BTreeSet, /// Send-side to transition a dataflow from read-only mode to read-write mode. /// /// All dataflows start in read-only mode. Only after receiving a @@ -2036,6 +2098,8 @@ impl CollectionState { compute_probe: None, logging: None, metrics, + owns_sink_frontier: false, + logged_stages: BTreeSet::new(), read_only_tx, read_only_rx, } @@ -2094,6 +2158,11 @@ impl CollectionState { } /// Return whether this collection is hydrated. + /// + /// This is the output-frontier reading, which folds in the write frontier and so reports + /// durability for a collection that sinks to persist. `observe_hydration` reports the + /// dataflow-progress reading instead. Both are wanted, and they differ for a materialized view + /// by the time its snapshot takes to reach persist. fn hydrated(&self) -> bool { match &self.reported_frontiers.output_frontier { ReportedFrontier::Reported(frontier) => PartialOrder::less_than(&self.as_of, frontier), @@ -2101,6 +2170,75 @@ impl CollectionState { } } + /// Log that this collection reached a lifecycle stage, unless it already reported it. + fn log_stage(&mut self, stage: LifecycleStage) { + if !self.logged_stages.insert(stage) { + return; + } + if let Some(logging) = &self.logging { + logging.log_lifecycle(stage); + } + } + + /// Observe whether this collection's dataflow has progressed past its as-of, and log the + /// `hydrated` stage the first time it has. + /// + /// The caller decides which frontier measures dataflow progress. See the comment at the call + /// site in `report_frontiers`. An empty as-of never hydrates, which is consistent with no + /// dataflow being created for one. + fn observe_hydration(&mut self, hydrated: bool) { + if hydrated { + self.log_stage(LifecycleStage::Hydrated); + } + } + + /// Observe whether this collection's sink has written past its as-of, and log the write + /// lifecycle stages it has reached. + /// + /// Only the worker that maintains the sink frontier reports these stages, which is what makes + /// them one observation per export rather than one per worker. + /// + /// Nothing is reported before the dataflow has hydrated. Until then the sink has produced no + /// output, so read-only mode is not holding anything back, and reporting a block there would + /// make `write_unblocked - hydrated` negative in the common case rather than zero. Gating here + /// also keeps the stages ordered against `written`, which can otherwise arrive first: + /// `apply_refresh` rounds a `REFRESH` materialized view's frontier up to the next refresh time + /// before the dataflow has computed anything, so its 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. + /// + /// NOTE: `written` is derived from the output shard's upper, which is a property of the shard + /// and not of this replica. The as-of is bounded to one step below that upper for a non-empty + /// storage export (`as_of_selection::apply_downstream_storage_constraints`), so for a shard + /// that already holds data the predicate is true from the moment the dataflow is installed. A + /// replica that may not write can therefore never be the one that advanced it, which is why + /// the stage is withheld while writes are blocked. Reporting it there would attribute another + /// writer's progress to this replica and put `written` before `write_unblocked`. + fn observe_writes(&mut self, written: bool) { + if !self.owns_sink_frontier || !self.logged_stages.contains(&LifecycleStage::Hydrated) { + return; + } + + let read_only = *self.read_only_rx.borrow(); + if read_only { + self.log_stage(LifecycleStage::WriteBlockedReadOnly); + return; + } + + if self + .logged_stages + .contains(&LifecycleStage::WriteBlockedReadOnly) + { + // Only report having been unblocked if we reported being blocked. A collection whose + // writes were allowed before we first observed it was never seen to wait. + self.log_stage(LifecycleStage::WriteUnblocked); + } + + if written { + self.log_stage(LifecycleStage::Written); + } + } + /// Allow writes for this collection. fn allow_writes(&self) { info!( diff --git a/src/compute/src/logging/compute.rs b/src/compute/src/logging/compute.rs index 065d51fceb240..47b76329f8377 100644 --- a/src/compute/src/logging/compute.rs +++ b/src/compute/src/logging/compute.rs @@ -57,6 +57,12 @@ pub struct Export { pub export_id: GlobalId, /// Timely worker index of the exporting dataflow. pub dataflow_index: usize, + /// The as-of of the exporting dataflow, unless it is the empty antichain. + /// + /// Every lifecycle stage is defined relative to the as-of, so the demux keeps it around to + /// report alongside the stages. Durations between stages are not comparable without it: the + /// same duration is a different amount of work for a recent as-of than for a far behind one. + pub as_of: Option, } /// The export for a global id was dropped. @@ -172,6 +178,54 @@ pub struct Hydration { pub export_id: GlobalId, } +/// An export reached a stage of its lifecycle. +#[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)] +pub struct Lifecycle { + /// Identifier of the export. + pub export_id: GlobalId, + /// The stage that was reached. + pub stage: LifecycleStage, +} + +/// A stage of an export's lifecycle. +/// +/// NOTE: Only the stages from [`LifecycleStage::Hydrated`] on are carried by a [`Lifecycle`] +/// event. `Installed` and `Started` are logged from the [`Export`] and [`HydrationStart`] events, +/// which already mark those moments for the hydration time relation. They are variants here so +/// that the stage vocabulary has a single definition. +#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug, Columnar)] +pub enum LifecycleStage { + /// The export's dataflow was installed, still suspended. + Installed, + /// The export's dataflow was unsuspended, so hydration work may begin. + Started, + /// The dataflow's own progress frontier passed its as-of. + Hydrated, + /// The export's sink may not write, because the dataflow is in read-only mode. + WriteBlockedReadOnly, + /// The export's sink may write. + WriteUnblocked, + /// The export's sink advanced the output shard's upper past the as-of. + Written, +} + +impl LifecycleStage { + /// The `event` and `reason` this stage is reported as. + /// + /// The reason is a closed vocabulary, so that "which exports are in this state, and why" stays + /// a filter over typed columns rather than a search through `details`. + fn columns(self) -> (&'static str, Option<&'static str>) { + match self { + Self::Installed => ("installed", None), + Self::Started => ("started", None), + Self::Hydrated => ("hydrated", None), + Self::WriteBlockedReadOnly => ("write_blocked", Some("read_only")), + Self::WriteUnblocked => ("write_unblocked", None), + Self::Written => ("written", None), + } + } +} + /// An operator's hydration status changed. #[derive(Debug, Clone, PartialOrd, PartialEq, Columnar)] pub struct OperatorHydration { @@ -238,6 +292,8 @@ pub enum ComputeEvent { HydrationStart(HydrationStart), /// A dataflow export was hydrated. Hydration(Hydration), + /// A dataflow export reached a stage of its lifecycle. + Lifecycle(Lifecycle), /// A dataflow operator's hydration status changed. OperatorHydration(OperatorHydration), /// An LIR operator was mapped to some particular dataflow operator. @@ -360,6 +416,8 @@ pub(super) fn construct<'scope>( let mut error_count_out = OutputBuilder::from(error_count_out); let (hydration_time_out, hydration_time) = demux.new_output(); let mut hydration_time_out = OutputBuilder::from(hydration_time_out); + let (lifecycle_out, lifecycle) = demux.new_output(); + let mut lifecycle_out = OutputBuilder::from(lifecycle_out); let (operator_hydration_status_out, operator_hydration_status) = demux.new_output(); let mut operator_hydration_status_out = OutputBuilder::from(operator_hydration_status_out); let (lir_mapping_out, lir_mapping) = demux.new_output(); @@ -380,6 +438,7 @@ pub(super) fn construct<'scope>( let mut arrangement_heap_allocations = arrangement_heap_allocations_out.activate(); let mut error_count = error_count_out.activate(); let mut hydration_time = hydration_time_out.activate(); + let mut lifecycle = lifecycle_out.activate(); let mut operator_hydration_status = operator_hydration_status_out.activate(); let mut lir_mapping = lir_mapping_out.activate(); let mut dataflow_global_ids = dataflow_global_ids_out.activate(); @@ -398,6 +457,7 @@ pub(super) fn construct<'scope>( arrangement_heap_size: arrangement_heap_size.session_with_builder(&cap), error_count: error_count.session_with_builder(&cap), hydration_time: hydration_time.session_with_builder(&cap), + lifecycle: lifecycle.session_with_builder(&cap), operator_hydration_status: operator_hydration_status .session_with_builder(&cap), lir_mapping: lir_mapping.session_with_builder(&cap), @@ -430,6 +490,7 @@ pub(super) fn construct<'scope>( (FrontierCurrent, frontier), (HydrationTime, hydration_time), (ImportFrontierCurrent, import_frontier), + (LifecycleEvent, lifecycle), (LirMapping, lir_mapping), (OperatorHydrationStatus, operator_hydration_status), (PeekCurrent, peek), @@ -494,6 +555,23 @@ fn epoch_offset_datum(offset: Duration) -> Datum<'static> { Datum::TimestampTz(timestamp) } +/// Pack the `details` payload reported alongside every lifecycle stage of an export. +/// +/// The as-of is rendered as a JSON string rather than a number. It is an `mz_timestamp`, which has +/// no faithful JSON number counterpart, and a string round-trips it exactly. +fn lifecycle_details(as_of: Option) -> Row { + let mut row = Row::default(); + let mut packer = row.packer(); + match as_of { + Some(ts) => { + let ts = ts.to_string(); + packer.push_dict([("as_of", Datum::String(&ts))]); + } + None => packer.push_dict([("as_of", Datum::JsonNull)]), + } + row +} + /// State maintained by the demux operator. struct DemuxState { /// The timely activations handle. @@ -540,6 +618,8 @@ struct DemuxState { peek_packer: PermutedRowPacker, /// A row packer for the hydration time output. hydration_time_packer: PermutedRowPacker, + /// A row packer for the lifecycle output. + lifecycle_packer: PermutedRowPacker, } impl DemuxState { @@ -567,6 +647,7 @@ impl DemuxState { frontier_packer: PermutedRowPacker::new(ComputeLog::FrontierCurrent), hydration_time_packer: PermutedRowPacker::new(ComputeLog::HydrationTime), import_frontier_packer: PermutedRowPacker::new(ComputeLog::ImportFrontierCurrent), + lifecycle_packer: PermutedRowPacker::new(ComputeLog::LifecycleEvent), lir_mapping_packer: PermutedRowPacker::new(ComputeLog::LirMapping), operator_hydration_status_packer: PermutedRowPacker::new( ComputeLog::OperatorHydrationStatus, @@ -663,6 +744,25 @@ impl DemuxState { ]) } + /// Pack a lifecycle update key-value for the given export ID and stage. + fn pack_lifecycle_update( + &mut self, + export_id: GlobalId, + stage: LifecycleStage, + occurred_at: Duration, + details: Datum<'_>, + ) -> (&RowRef, &RowRef) { + let (event, reason) = stage.columns(); + self.lifecycle_packer.pack_slice(&[ + make_string_datum(export_id, &mut self.scratch_string_a), + Datum::UInt64(u64::cast_from(self.worker_id)), + Datum::String(event), + epoch_offset_datum(occurred_at), + reason.map_or(Datum::Null, Datum::String), + details, + ]) + } + /// Pack an import frontier update key-value for the given export ID and dataflow index. fn pack_import_frontier_update( &mut self, @@ -799,10 +899,18 @@ struct ExportState { hydration_timestamps: HydrationTimestamps, /// Hydration status of operators feeding this export. operator_hydration: BTreeMap, + /// The as-of of the dataflow maintaining this export. + as_of: Option, + /// The lifecycle rows logged for this export so far. + /// + /// The lifecycle relation is append-only for the life of an export, so the rows are kept to + /// retract them when it is dropped. Re-deriving them at drop time would risk drifting from + /// what was inserted. + lifecycle_rows: Vec<(Row, Row)>, } impl ExportState { - fn new(dataflow_index: usize, installed_at: Duration) -> Self { + fn new(dataflow_index: usize, installed_at: Duration, as_of: Option) -> Self { Self { dataflow_index, error_count: Diff::ZERO, @@ -814,6 +922,8 @@ impl ExportState { hydrated_at: None, }, operator_hydration: BTreeMap::new(), + as_of, + lifecycle_rows: Vec::new(), } } } @@ -837,6 +947,7 @@ struct DemuxOutput<'a, 'b> { arrangement_heap_capacity: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>, arrangement_heap_size: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>, hydration_time: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>, + lifecycle: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>, operator_hydration_status: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>, error_count: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>, lir_mapping: OutputSessionColumnar<'a, 'b, Update<(Row, Row)>>, @@ -888,6 +999,7 @@ impl DemuxHandler<'_, '_, '_> { ErrorCount(error_count) => self.handle_error_count(error_count), HydrationStart(hydration) => self.handle_hydration_start(hydration), Hydration(hydration) => self.handle_hydration(hydration), + Lifecycle(lifecycle) => self.handle_lifecycle(lifecycle), OperatorHydration(hydration) => self.handle_operator_hydration(hydration), LirMapping(mapping) => self.handle_lir_mapping(mapping), DataflowGlobal(global) => self.handle_dataflow_global(global), @@ -899,6 +1011,7 @@ impl DemuxHandler<'_, '_, '_> { ExportReference { export_id, dataflow_index, + as_of, }: Ref<'_, Export>, ) { let export_id = Columnar::into_owned(export_id); @@ -910,10 +1023,11 @@ impl DemuxHandler<'_, '_, '_> { // then only delays when an update becomes visible, rather than skewing recorded instants. let installed_at = self.time; - let existing = self - .state - .exports - .insert(export_id, ExportState::new(dataflow_index, installed_at)); + let as_of = Option::::into_owned(as_of); + let existing = self.state.exports.insert( + export_id, + ExportState::new(dataflow_index, installed_at, as_of), + ); if existing.is_some() { error!(%export_id, "export already registered"); } @@ -928,6 +1042,8 @@ impl DemuxHandler<'_, '_, '_> { .state .pack_hydration_time_update(export_id, None, ×tamps); self.output.hydration_time.give((datum, ts, Diff::ONE)); + + self.log_lifecycle(export_id, LifecycleStage::Installed); } fn handle_export_dropped( @@ -964,6 +1080,13 @@ impl DemuxHandler<'_, '_, '_> { .hydration_time .give((datum, ts, Diff::MINUS_ONE)); + // Remove lifecycle logging for this export. + for (key, value) in &export.lifecycle_rows { + self.output + .lifecycle + .give(((&**key, &**value), ts, Diff::MINUS_ONE)); + } + // Remove operator hydration logging for this export. for (lir_id, hydrated) in export.operator_hydration { let datum = self @@ -1078,6 +1201,8 @@ impl DemuxHandler<'_, '_, '_> { .state .pack_hydration_time_update(export_id, time_ns, &new_timestamps); self.output.hydration_time.give((insertion, ts, Diff::ONE)); + + self.log_lifecycle(export_id, LifecycleStage::Started); } fn handle_hydration(&mut self, HydrationReference { export_id }: Ref<'_, Hydration>) { @@ -1111,7 +1236,8 @@ impl DemuxHandler<'_, '_, '_> { // hydrated_at` total and reports the queueing interval as zero. Stamping `hydrated_at` // instead would invert it, charging the whole life to queueing and reporting zero // hydration time for a dataflow that only ever hydrated. - if export.hydration_timestamps.started_at.is_none() { + let backfilled_start = export.hydration_timestamps.started_at.is_none(); + if backfilled_start { export.hydration_timestamps.started_at = Some(export.hydration_timestamps.installed_at); } let new_timestamps = export.hydration_timestamps; @@ -1126,6 +1252,72 @@ impl DemuxHandler<'_, '_, '_> { self.state .pack_hydration_time_update(export_id, Some(nanos), &new_timestamps); self.output.hydration_time.give((insertion, ts, Diff::ONE)); + + // The lifecycle log needs the same back-fill. A `Schedule` that arrives after hydration is + // absorbed by the guard in `handle_hydration_start`, so this is the only chance to report + // the stage, and without it the export would report `hydrated` with no `started`. + // + // Stamp it from `installed_at`, not from the current event time, for the same reason the + // timestamps above do. A dataflow that hydrated before it was ever scheduled did not + // queue, so reporting `started` at the hydration instant would charge its whole life to + // queueing and report roughly zero hydration, and the two relations would disagree about + // the same export. + if backfilled_start { + self.log_lifecycle_at( + export_id, + LifecycleStage::Started, + new_timestamps.installed_at, + ); + } + } + + /// Log an export having reached a lifecycle stage, and remember the row so that it can be + /// retracted when the export is dropped. + fn log_lifecycle(&mut self, export_id: GlobalId, stage: LifecycleStage) { + // Stamp the event time rather than `ts`, as in `handle_export`. + let occurred_at = self.time; + self.log_lifecycle_at(export_id, stage, occurred_at); + } + + /// As [`Self::log_lifecycle`], for a stage whose instant is not the current event time. + fn log_lifecycle_at( + &mut self, + export_id: GlobalId, + stage: LifecycleStage, + occurred_at: Duration, + ) { + let ts = self.ts(); + + let Some(as_of) = self.state.exports.get(&export_id).map(|e| e.as_of) else { + error!(%export_id, ?stage, "lifecycle event for unknown export"); + return; + }; + + let details = lifecycle_details(as_of); + let update = { + let (key, value) = self.state.pack_lifecycle_update( + export_id, + stage, + occurred_at, + details.unpack_first(), + ); + (key.to_owned(), value.to_owned()) + }; + self.output + .lifecycle + .give(((&*update.0, &*update.1), ts, Diff::ONE)); + + let export = self + .state + .exports + .get_mut(&export_id) + .expect("checked above"); + export.lifecycle_rows.push(update); + } + + fn handle_lifecycle(&mut self, LifecycleReference { export_id, stage }: Ref<'_, Lifecycle>) { + let export_id = Columnar::into_owned(export_id); + self.log_lifecycle(export_id, stage); } fn handle_operator_hydration( @@ -1454,11 +1646,13 @@ impl CollectionLogging { export_id: GlobalId, logger: Logger, dataflow_index: usize, + as_of: Option, import_ids: impl Iterator, ) -> Self { logger.log(&ComputeEvent::Export(Export { export_id, dataflow_index, + as_of, })); let mut self_ = Self { @@ -1544,6 +1738,17 @@ impl CollectionLogging { })); } + /// Record that the collection reached a stage of its lifecycle. + /// + /// Callers must not report a stage twice, since the lifecycle relation is append-only and the + /// demux does not deduplicate. + pub fn log_lifecycle(&self, stage: LifecycleStage) { + self.logger.log(&ComputeEvent::Lifecycle(Lifecycle { + export_id: self.export_id, + stage, + })); + } + /// Set the collection as hydrated. pub fn set_hydrated(&self) { self.logger.log(&ComputeEvent::Hydration(Hydration { diff --git a/src/compute/src/sink/materialized_view.rs b/src/compute/src/sink/materialized_view.rs index e5d1a4064c1ba..889bc768a1512 100644 --- a/src/compute/src/sink/materialized_view.rs +++ b/src/compute/src/sink/materialized_view.rs @@ -259,6 +259,7 @@ where } let scope = ok_collection.scope(); + let owns_sink_frontier = scope.index() == frontier_owner(sink_id, scope.peers()); let desired = OkErr::new(ok_collection.inner, err_collection.inner); // Read back the persist shard. @@ -296,6 +297,7 @@ where // Report sink frontier updates to the `ComputeState`. let collection = compute_state.expect_collection_mut(sink_id); collection.sink_write_frontier = Some(sink_frontier); + collection.owns_sink_frontier = owns_sink_frontier; Rc::new((persist_token, mint_token, write_token, append_token)) } @@ -471,6 +473,16 @@ impl std::fmt::Debug for BatchDescription { } } +/// The worker that maintains a sink's shared write frontier. +/// +/// The `mint` operator tracks the output shard's upper on this worker alone and clears the shared +/// frontier on all the others, so only this worker's copy carries write progress. Anything that +/// reads the shared frontier as a measure of writing, rather than as an input to the +/// controller-visible meet, must agree with this election. +pub(super) fn frontier_owner(sink_id: GlobalId, peers: usize) -> usize { + usize::cast_from(sink_id.hashed()) % peers +} + /// Construct a name for the given sub-operator. pub(super) fn operator_name(sink_id: GlobalId, sub_operator: &str) -> String { format!("mv_sink({sink_id})::{sub_operator}") @@ -505,7 +517,7 @@ mod mint { let worker_count = scope.peers(); // Determine the active worker for the mint operator. - let active_worker_id = usize::cast_from(sink_id.hashed()) % scope.peers(); + let active_worker_id = super::frontier_owner(sink_id, scope.peers()); let sink_frontier = Rc::new(RefCell::new(Antichain::from_elem(Timestamp::MIN))); let shared_frontier = Rc::clone(&sink_frontier); diff --git a/src/compute/src/sink/materialized_view_v2.rs b/src/compute/src/sink/materialized_view_v2.rs index 5408d63142025..758e8c1305798 100644 --- a/src/compute/src/sink/materialized_view_v2.rs +++ b/src/compute/src/sink/materialized_view_v2.rs @@ -94,6 +94,8 @@ pub(super) fn persist_sink<'s>( read_only_rx: watch::Receiver, ) -> Rc { let scope = ok_collection.scope(); + let owns_sink_frontier = + scope.index() == super::materialized_view::frontier_owner(sink_id, scope.peers()); let desired = OkErr::new(ok_collection.inner, err_collection.inner); // Read back the persist shard. @@ -136,6 +138,7 @@ pub(super) fn persist_sink<'s>( // Report sink frontier updates to the `ComputeState`. let collection = compute_state.expect_collection_mut(sink_id); collection.sink_write_frontier = Some(sink_frontier); + collection.owns_sink_frontier = owns_sink_frontier; Rc::new(persist_token) } @@ -165,7 +168,8 @@ mod mint { let worker_count = scope.peers(); // Determine the active worker for the mint operator. - let active_worker_id = usize::cast_from(sink_id.hashed()) % scope.peers(); + let active_worker_id = + crate::sink::materialized_view::frontier_owner(sink_id, scope.peers()); let sink_frontier = Rc::new(RefCell::new(Antichain::from_elem(Timestamp::MIN))); let shared_frontier = Rc::clone(&sink_frontier); diff --git a/src/pgrepr-consts/src/oid.rs b/src/pgrepr-consts/src/oid.rs index e906f223f34bd..45273565bcffa 100644 --- a/src/pgrepr-consts/src/oid.rs +++ b/src/pgrepr-consts/src/oid.rs @@ -826,3 +826,4 @@ pub const VIEW_MZ_OBJECT_GRAPH_EDGES_OID: u32 = 17116; pub const INDEX_MZ_OBJECT_GRAPH_EDGES_IND_OID: u32 = 17117; pub const VIEW_MZ_BUILTIN_TABLES_OID: u32 = 17118; pub const VIEW_MZ_BUILTIN_VIEWS_OID: u32 = 17119; +pub const LOG_MZ_COMPUTE_LIFECYCLE_EVENTS_PER_WORKER_OID: u32 = 17120; diff --git a/test/sqllogictest/autogenerated/mz_introspection.slt b/test/sqllogictest/autogenerated/mz_introspection.slt index a914ba4344620..65a3cf3769f36 100644 --- a/test/sqllogictest/autogenerated/mz_introspection.slt +++ b/test/sqllogictest/autogenerated/mz_introspection.slt @@ -282,6 +282,7 @@ mz_compute_frontiers_per_worker mz_compute_hydration_times_per_worker mz_compute_import_frontiers mz_compute_import_frontiers_per_worker +mz_compute_lifecycle_events_per_worker mz_compute_lir_mapping_per_worker mz_compute_operator_durations_histogram mz_compute_operator_durations_histogram_per_worker diff --git a/test/sqllogictest/catalog_server_explain.slt b/test/sqllogictest/catalog_server_explain.slt index 6a2ed84d1f888..9060551378e52 100644 --- a/test/sqllogictest/catalog_server_explain.slt +++ b/test/sqllogictest/catalog_server_explain.slt @@ -4967,7 +4967,7 @@ mz_catalog.mz_indexes: Filter: ("2" = (#1 ->> "object_type")) AND ("mz_introspection" = (#1 ->> "schema_name")) AND ((#1 ->> "object_name")) IS NOT NULL AND ("GidMapping" = #2) →Read mz_internal.mz_catalog_raw →Arrange (#0{log_name}) - →Constant (32 rows) + →Constant (33 rows) Source mz_internal.mz_catalog_raw project=(#0..=#2) @@ -5373,7 +5373,7 @@ mz_catalog.mz_sources: Project: #4, #0, #5, #1, #2, #6..=#12, #3, #13, #14 Map: null, null, null, null, null, null, "s1", null, null →Arrange (#1{schema_name}, #2{name}) - →Constant (56 rows) + →Constant (57 rows) →Arrange (#0{schema_name}) (#0{schema_name}, #1{name}) →Fused with Child Map/Filter/Project Project: #4, #3, #5 @@ -8202,7 +8202,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_builtin_sources"; ---- Explained Query (fast path): - →Constant (56 rows) + →Constant (57 rows) Target cluster: mz_catalog_server @@ -10204,7 +10204,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_ontology_entity_types"; ---- Explained Query (fast path): - →Constant (134 rows) + →Constant (135 rows) Target cluster: mz_catalog_server @@ -10214,7 +10214,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_ontology_link_types"; ---- Explained Query (fast path): - →Constant (173 rows) + →Constant (174 rows) Target cluster: mz_catalog_server @@ -10228,7 +10228,7 @@ Explained Query: cte l0 = →Differential Join %1:mz_schemas[#0{id}] » %2:mz_objects[#2{schema_id}] » %0[#0{schema_name}, #1{table_name}] » %3:mz_columns[#0{id}] →Arrange (#0{schema_name}, #1{table_name}) - →Constant (134 rows) + →Constant (135 rows) →Arrange (#0{id}) →Fused with Child Map/Filter/Project Project: #1, #3 @@ -10281,7 +10281,7 @@ Explained Query: →Differential Join %0:l4[#0{entity_name}, #1{name}] » %1[#0{entity_name}, #1{column_name}] →Arranged l4 →Arrange (#0{entity_name}, #1{column_name}) - →Constant (272 rows) + →Constant (273 rows) →Return →Union →Map/Filter/Project diff --git a/test/sqllogictest/cluster.slt b/test/sqllogictest/cluster.slt index 10cb6649b6d62..ca9470904888d 100644 --- a/test/sqllogictest/cluster.slt +++ b/test/sqllogictest/cluster.slt @@ -215,6 +215,12 @@ bar mz_compute_hydration_times_per_worker mz_compute_hydration_times_per_worke bar mz_compute_import_frontiers_per_worker mz_compute_import_frontiers_per_worker_u7_primary_idx 1 export_id NULL false bar mz_compute_import_frontiers_per_worker mz_compute_import_frontiers_per_worker_u7_primary_idx 2 import_id NULL false bar mz_compute_import_frontiers_per_worker mz_compute_import_frontiers_per_worker_u7_primary_idx 3 worker_id NULL false +bar mz_compute_lifecycle_events_per_worker mz_compute_lifecycle_events_per_worker_u7_primary_idx 1 export_id NULL false +bar mz_compute_lifecycle_events_per_worker mz_compute_lifecycle_events_per_worker_u7_primary_idx 2 worker_id NULL false +bar mz_compute_lifecycle_events_per_worker mz_compute_lifecycle_events_per_worker_u7_primary_idx 3 event NULL false +bar mz_compute_lifecycle_events_per_worker mz_compute_lifecycle_events_per_worker_u7_primary_idx 4 occurred_at NULL false +bar mz_compute_lifecycle_events_per_worker mz_compute_lifecycle_events_per_worker_u7_primary_idx 5 reason NULL true +bar mz_compute_lifecycle_events_per_worker mz_compute_lifecycle_events_per_worker_u7_primary_idx 6 details NULL true bar mz_compute_lir_mapping_per_worker mz_compute_lir_mapping_per_worker_u7_primary_idx 1 global_id NULL false bar mz_compute_lir_mapping_per_worker mz_compute_lir_mapping_per_worker_u7_primary_idx 2 lir_id NULL false bar mz_compute_lir_mapping_per_worker mz_compute_lir_mapping_per_worker_u7_primary_idx 3 worker_id NULL false @@ -407,7 +413,7 @@ DROP CLUSTER foo, foo2, foo3, foo4 CASCADE query I SELECT COUNT(name) FROM mz_indexes WHERE cluster_id = 'u1'; ---- -32 +33 query I SELECT COUNT(name) FROM mz_indexes WHERE cluster_id <> 'u1' AND cluster_id NOT LIKE 's%'; @@ -420,7 +426,7 @@ CREATE CLUSTER test REPLICAS (foo (SIZE 'scale=1,workers=1')); query I SELECT COUNT(name) FROM mz_indexes; ---- -305 +312 statement ok DROP CLUSTER test CASCADE @@ -428,7 +434,7 @@ DROP CLUSTER test CASCADE query T SELECT COUNT(name) FROM mz_indexes; ---- -273 +279 simple conn=mz_system,user=mz_system ALTER CLUSTER quickstart OWNER TO materialize diff --git a/test/sqllogictest/cockroach/srfs.slt b/test/sqllogictest/cockroach/srfs.slt index 10592810e8f52..7fa9709720e7e 100644 --- a/test/sqllogictest/cockroach/srfs.slt +++ b/test/sqllogictest/cockroach/srfs.slt @@ -1258,6 +1258,42 @@ mz_compute_import_frontiers_per_worker 3 mz_compute_import_frontiers_per_worker 3 mz_compute_import_frontiers_per_worker 3 mz_compute_import_frontiers_per_worker 3 +mz_compute_lifecycle_events_per_worker 1 +mz_compute_lifecycle_events_per_worker 1 +mz_compute_lifecycle_events_per_worker 1 +mz_compute_lifecycle_events_per_worker 1 +mz_compute_lifecycle_events_per_worker 1 +mz_compute_lifecycle_events_per_worker 1 +mz_compute_lifecycle_events_per_worker 2 +mz_compute_lifecycle_events_per_worker 2 +mz_compute_lifecycle_events_per_worker 2 +mz_compute_lifecycle_events_per_worker 2 +mz_compute_lifecycle_events_per_worker 2 +mz_compute_lifecycle_events_per_worker 2 +mz_compute_lifecycle_events_per_worker 3 +mz_compute_lifecycle_events_per_worker 3 +mz_compute_lifecycle_events_per_worker 3 +mz_compute_lifecycle_events_per_worker 3 +mz_compute_lifecycle_events_per_worker 3 +mz_compute_lifecycle_events_per_worker 3 +mz_compute_lifecycle_events_per_worker 4 +mz_compute_lifecycle_events_per_worker 4 +mz_compute_lifecycle_events_per_worker 4 +mz_compute_lifecycle_events_per_worker 4 +mz_compute_lifecycle_events_per_worker 4 +mz_compute_lifecycle_events_per_worker 4 +mz_compute_lifecycle_events_per_worker 5 +mz_compute_lifecycle_events_per_worker 5 +mz_compute_lifecycle_events_per_worker 5 +mz_compute_lifecycle_events_per_worker 5 +mz_compute_lifecycle_events_per_worker 5 +mz_compute_lifecycle_events_per_worker 5 +mz_compute_lifecycle_events_per_worker 6 +mz_compute_lifecycle_events_per_worker 6 +mz_compute_lifecycle_events_per_worker 6 +mz_compute_lifecycle_events_per_worker 6 +mz_compute_lifecycle_events_per_worker 6 +mz_compute_lifecycle_events_per_worker 6 mz_compute_lir_mapping_per_worker 1 mz_compute_lir_mapping_per_worker 1 mz_compute_lir_mapping_per_worker 1 diff --git a/test/sqllogictest/distinct_arrangements.slt b/test/sqllogictest/distinct_arrangements.slt index 522db9cc25a15..b1f93ad23ece1 100644 --- a/test/sqllogictest/distinct_arrangements.slt +++ b/test/sqllogictest/distinct_arrangements.slt @@ -1112,6 +1112,7 @@ Arrange Compute(ErrorCount) Arrange Compute(FrontierCurrent) Arrange Compute(HydrationTime) Arrange Compute(ImportFrontierCurrent) +Arrange Compute(LifecycleEvent) Arrange Compute(LirMapping) Arrange Compute(OperatorHydrationStatus) Arrange Compute(PeekCurrent) diff --git a/test/sqllogictest/information_schema_tables.slt b/test/sqllogictest/information_schema_tables.slt index 6aa277b21008a..51377351f5b35 100644 --- a/test/sqllogictest/information_schema_tables.slt +++ b/test/sqllogictest/information_schema_tables.slt @@ -981,6 +981,10 @@ mz_compute_import_frontiers_per_worker SOURCE materialize mz_introspection +mz_compute_lifecycle_events_per_worker +SOURCE +materialize +mz_introspection mz_compute_lir_mapping_per_worker SOURCE materialize diff --git a/test/sqllogictest/introspection/relations.slt b/test/sqllogictest/introspection/relations.slt index b8acec4390bb0..fa4c9266564b1 100644 --- a/test/sqllogictest/introspection/relations.slt +++ b/test/sqllogictest/introspection/relations.slt @@ -124,6 +124,7 @@ Arrange␠Compute(ErrorCount) ArrangementSize alloc::vec::Vec)>>>> Arrange␠Compute(HydrationTime) ArrangementSize alloc::vec::Vec)>>>> Arrange␠Compute(ImportFrontierCurrent) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(LifecycleEvent) ArrangementSize alloc::vec::Vec)>>>> Arrange␠Compute(LirMapping) ArrangementSize alloc::vec::Vec)>>>> Arrange␠Compute(OperatorHydrationStatus) ArrangementSize alloc::vec::Vec)>>>> Arrange␠Compute(PeekCurrent) ArrangementSize alloc::vec::Vec)>>>> @@ -156,6 +157,7 @@ Compute␠Logging␠Demux Arrange␠Compute(ErrorCount) mz_timely_util::column Compute␠Logging␠Demux Arrange␠Compute(FrontierCurrent) mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Compute␠Logging␠Demux Arrange␠Compute(HydrationTime) mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Compute␠Logging␠Demux Arrange␠Compute(ImportFrontierCurrent) mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> +Compute␠Logging␠Demux Arrange␠Compute(LifecycleEvent) mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Compute␠Logging␠Demux Arrange␠Compute(LirMapping) mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Compute␠Logging␠Demux Arrange␠Compute(OperatorHydrationStatus) mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Compute␠Logging␠Demux Arrange␠Compute(PeekCurrent) mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> @@ -245,8 +247,8 @@ GROUP BY type; 1 mz_timely_util::columnar::Column<(core::time::Duration,␠(usize,␠alloc::vec::Vec<(usize,␠usize,␠bool,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)>))> 1 mz_timely_util::columnar::Column<(core::time::Duration,␠mz_compute::logging::compute::ComputeEvent)> 3 alloc::vec::Vec<(core::time::Duration,␠timely::logging::TimelyEvent)> -32 alloc::vec::Vec)>>>> -32 mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> +33 alloc::vec::Vec)>>>> +33 mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> 4 alloc::vec::Vec<((mz_compute::logging::timely::MessageDatum,␠()),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> 4 alloc::vec::Vec)>>> 8 alloc::vec::Vec<((usize,␠()),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> diff --git a/test/sqllogictest/mz_catalog_server_index_accounting.slt b/test/sqllogictest/mz_catalog_server_index_accounting.slt index 6e9b993db5d58..a1d7d1bd17729 100644 --- a/test/sqllogictest/mz_catalog_server_index_accounting.slt +++ b/test/sqllogictest/mz_catalog_server_index_accounting.slt @@ -37,108 +37,109 @@ mz_arrangement_heap_capacity_raw_s2_primary_idx CREATE␠INDEX␠"mz_arrangemen mz_arrangement_heap_size_raw_s2_primary_idx CREATE␠INDEX␠"mz_arrangement_heap_size_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_arrangement_heap_size_raw"␠("operator_id",␠"worker_id") mz_arrangement_records_raw_s2_primary_idx CREATE␠INDEX␠"mz_arrangement_records_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_arrangement_records_raw"␠("operator_id",␠"worker_id") mz_arrangement_sharing_raw_s2_primary_idx CREATE␠INDEX␠"mz_arrangement_sharing_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_arrangement_sharing_raw"␠("operator_id",␠"worker_id") -mz_cluster_auto_scaling_strategies_ind CREATE␠INDEX␠"mz_cluster_auto_scaling_strategies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s520␠AS␠"mz_internal"."mz_cluster_auto_scaling_strategies"]␠("cluster_id") -mz_cluster_deployment_lineage_ind CREATE␠INDEX␠"mz_cluster_deployment_lineage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s768␠AS␠"mz_internal"."mz_cluster_deployment_lineage"]␠("cluster_id") +mz_cluster_auto_scaling_strategies_ind CREATE␠INDEX␠"mz_cluster_auto_scaling_strategies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s521␠AS␠"mz_internal"."mz_cluster_auto_scaling_strategies"]␠("cluster_id") +mz_cluster_deployment_lineage_ind CREATE␠INDEX␠"mz_cluster_deployment_lineage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s769␠AS␠"mz_internal"."mz_cluster_deployment_lineage"]␠("cluster_id") mz_cluster_prometheus_metrics_s2_primary_idx CREATE␠INDEX␠"mz_cluster_prometheus_metrics_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_cluster_prometheus_metrics"␠("process_id",␠"metric_name",␠"labels") -mz_cluster_reconfigurations_ind CREATE␠INDEX␠"mz_cluster_reconfigurations_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s519␠AS␠"mz_internal"."mz_cluster_reconfigurations"]␠("cluster_id") -mz_cluster_replica_frontiers_ind CREATE␠INDEX␠"mz_cluster_replica_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s762␠AS␠"mz_catalog"."mz_cluster_replica_frontiers"]␠("object_id") -mz_cluster_replica_history_ind CREATE␠INDEX␠"mz_cluster_replica_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s617␠AS␠"mz_internal"."mz_cluster_replica_history"]␠("dropped_at") -mz_cluster_replica_metrics_history_ind CREATE␠INDEX␠"mz_cluster_replica_metrics_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s525␠AS␠"mz_internal"."mz_cluster_replica_metrics_history"]␠("replica_id") -mz_cluster_replica_metrics_ind CREATE␠INDEX␠"mz_cluster_replica_metrics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s526␠AS␠"mz_internal"."mz_cluster_replica_metrics"]␠("replica_id") -mz_cluster_replica_name_history_ind CREATE␠INDEX␠"mz_cluster_replica_name_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s618␠AS␠"mz_internal"."mz_cluster_replica_name_history"]␠("id") -mz_cluster_replica_size_internal_ind CREATE␠INDEX␠"mz_cluster_replica_size_internal_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s515␠AS␠"mz_internal"."mz_cluster_replica_size_internal"]␠("size") -mz_cluster_replica_sizes_ind CREATE␠INDEX␠"mz_cluster_replica_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s514␠AS␠"mz_catalog"."mz_cluster_replica_sizes"]␠("size") -mz_cluster_replica_status_history_ind CREATE␠INDEX␠"mz_cluster_replica_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s527␠AS␠"mz_internal"."mz_cluster_replica_status_history"]␠("replica_id") -mz_cluster_replica_statuses_ind CREATE␠INDEX␠"mz_cluster_replica_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s528␠AS␠"mz_internal"."mz_cluster_replica_statuses"]␠("replica_id") -mz_cluster_replicas_ind CREATE␠INDEX␠"mz_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s524␠AS␠"mz_catalog"."mz_cluster_replicas"]␠("id") -mz_clusters_ind CREATE␠INDEX␠"mz_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s516␠AS␠"mz_catalog"."mz_clusters"]␠("id") -mz_columns_ind CREATE␠INDEX␠"mz_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s487␠AS␠"mz_catalog"."mz_columns"]␠("name") -mz_comments_ind CREATE␠INDEX␠"mz_comments_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s543␠AS␠"mz_internal"."mz_comments"]␠("id") +mz_cluster_reconfigurations_ind CREATE␠INDEX␠"mz_cluster_reconfigurations_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s520␠AS␠"mz_internal"."mz_cluster_reconfigurations"]␠("cluster_id") +mz_cluster_replica_frontiers_ind CREATE␠INDEX␠"mz_cluster_replica_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s763␠AS␠"mz_catalog"."mz_cluster_replica_frontiers"]␠("object_id") +mz_cluster_replica_history_ind CREATE␠INDEX␠"mz_cluster_replica_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s618␠AS␠"mz_internal"."mz_cluster_replica_history"]␠("dropped_at") +mz_cluster_replica_metrics_history_ind CREATE␠INDEX␠"mz_cluster_replica_metrics_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s526␠AS␠"mz_internal"."mz_cluster_replica_metrics_history"]␠("replica_id") +mz_cluster_replica_metrics_ind CREATE␠INDEX␠"mz_cluster_replica_metrics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s527␠AS␠"mz_internal"."mz_cluster_replica_metrics"]␠("replica_id") +mz_cluster_replica_name_history_ind CREATE␠INDEX␠"mz_cluster_replica_name_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s619␠AS␠"mz_internal"."mz_cluster_replica_name_history"]␠("id") +mz_cluster_replica_size_internal_ind CREATE␠INDEX␠"mz_cluster_replica_size_internal_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s516␠AS␠"mz_internal"."mz_cluster_replica_size_internal"]␠("size") +mz_cluster_replica_sizes_ind CREATE␠INDEX␠"mz_cluster_replica_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s515␠AS␠"mz_catalog"."mz_cluster_replica_sizes"]␠("size") +mz_cluster_replica_status_history_ind CREATE␠INDEX␠"mz_cluster_replica_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s528␠AS␠"mz_internal"."mz_cluster_replica_status_history"]␠("replica_id") +mz_cluster_replica_statuses_ind CREATE␠INDEX␠"mz_cluster_replica_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s529␠AS␠"mz_internal"."mz_cluster_replica_statuses"]␠("replica_id") +mz_cluster_replicas_ind CREATE␠INDEX␠"mz_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s525␠AS␠"mz_catalog"."mz_cluster_replicas"]␠("id") +mz_clusters_ind CREATE␠INDEX␠"mz_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s517␠AS␠"mz_catalog"."mz_clusters"]␠("id") +mz_columns_ind CREATE␠INDEX␠"mz_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s488␠AS␠"mz_catalog"."mz_columns"]␠("name") +mz_comments_ind CREATE␠INDEX␠"mz_comments_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s544␠AS␠"mz_internal"."mz_comments"]␠("id") mz_compute_dataflow_global_ids_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_dataflow_global_ids_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_dataflow_global_ids_per_worker"␠("id",␠"worker_id",␠"global_id") -mz_compute_dependencies_ind CREATE␠INDEX␠"mz_compute_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s743␠AS␠"mz_internal"."mz_compute_dependencies"]␠("dependency_id") +mz_compute_dependencies_ind CREATE␠INDEX␠"mz_compute_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s744␠AS␠"mz_internal"."mz_compute_dependencies"]␠("dependency_id") mz_compute_error_counts_raw_s2_primary_idx CREATE␠INDEX␠"mz_compute_error_counts_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_error_counts_raw"␠("export_id",␠"worker_id") mz_compute_exports_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_exports_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_exports_per_worker"␠("export_id",␠"worker_id") mz_compute_frontiers_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_frontiers_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_frontiers_per_worker"␠("export_id",␠"worker_id") -mz_compute_hydration_times_ind CREATE␠INDEX␠"mz_compute_hydration_times_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s753␠AS␠"mz_internal"."mz_compute_hydration_times"]␠("replica_id") +mz_compute_hydration_times_ind CREATE␠INDEX␠"mz_compute_hydration_times_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s754␠AS␠"mz_internal"."mz_compute_hydration_times"]␠("replica_id") mz_compute_hydration_times_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_hydration_times_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_hydration_times_per_worker"␠("export_id",␠"worker_id") mz_compute_import_frontiers_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_import_frontiers_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_import_frontiers_per_worker"␠("export_id",␠"import_id",␠"worker_id") +mz_compute_lifecycle_events_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_lifecycle_events_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_lifecycle_events_per_worker"␠("export_id",␠"worker_id",␠"event",␠"occurred_at",␠"reason",␠"details") mz_compute_lir_mapping_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_lir_mapping_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_lir_mapping_per_worker"␠("global_id",␠"lir_id",␠"worker_id") mz_compute_operator_durations_histogram_raw_s2_primary_idx CREATE␠INDEX␠"mz_compute_operator_durations_histogram_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_operator_durations_histogram_raw"␠("id",␠"worker_id",␠"duration_ns") mz_compute_operator_hydration_statuses_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_operator_hydration_statuses_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_operator_hydration_statuses_per_worker"␠("export_id",␠"lir_id",␠"worker_id") -mz_connections_ind CREATE␠INDEX␠"mz_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s522␠AS␠"mz_catalog"."mz_connections"]␠("schema_id") -mz_console_cluster_utilization_overview_24h_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_24h_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s749␠AS␠"mz_internal"."mz_console_cluster_utilization_overview_24h"]␠("cluster_id") -mz_console_cluster_utilization_overview_3h_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_3h_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s748␠AS␠"mz_internal"."mz_console_cluster_utilization_overview_3h"]␠("cluster_id") -mz_console_cluster_utilization_overview_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s747␠AS␠"mz_internal"."mz_console_cluster_utilization_overview"]␠("cluster_id") -mz_databases_ind CREATE␠INDEX␠"mz_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s485␠AS␠"mz_catalog"."mz_databases"]␠("name") +mz_connections_ind CREATE␠INDEX␠"mz_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s523␠AS␠"mz_catalog"."mz_connections"]␠("schema_id") +mz_console_cluster_utilization_overview_24h_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_24h_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s750␠AS␠"mz_internal"."mz_console_cluster_utilization_overview_24h"]␠("cluster_id") +mz_console_cluster_utilization_overview_3h_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_3h_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s749␠AS␠"mz_internal"."mz_console_cluster_utilization_overview_3h"]␠("cluster_id") +mz_console_cluster_utilization_overview_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s748␠AS␠"mz_internal"."mz_console_cluster_utilization_overview"]␠("cluster_id") +mz_databases_ind CREATE␠INDEX␠"mz_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s486␠AS␠"mz_catalog"."mz_databases"]␠("name") mz_dataflow_addresses_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_dataflow_addresses_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_addresses_per_worker"␠("id",␠"worker_id") mz_dataflow_channels_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_dataflow_channels_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_channels_per_worker"␠("id",␠"worker_id") mz_dataflow_operator_reachability_raw_s2_primary_idx CREATE␠INDEX␠"mz_dataflow_operator_reachability_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_operator_reachability_raw"␠("id",␠"worker_id",␠"source",␠"port",␠"update_type",␠"time") mz_dataflow_operators_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_dataflow_operators_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_operators_per_worker"␠("id",␠"worker_id") -mz_frontiers_ind CREATE␠INDEX␠"mz_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s734␠AS␠"mz_internal"."mz_frontiers"]␠("object_id") -mz_hydration_statuses_ind CREATE␠INDEX␠"mz_hydration_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s764␠AS␠"mz_internal"."mz_hydration_statuses"]␠("object_id",␠"replica_id") -mz_indexes_ind CREATE␠INDEX␠"mz_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s488␠AS␠"mz_catalog"."mz_indexes"]␠("id") -mz_kafka_sources_ind CREATE␠INDEX␠"mz_kafka_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s482␠AS␠"mz_catalog"."mz_kafka_sources"]␠("id") -mz_materialized_views_ind CREATE␠INDEX␠"mz_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s546␠AS␠"mz_catalog"."mz_materialized_views"]␠("id") +mz_frontiers_ind CREATE␠INDEX␠"mz_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s735␠AS␠"mz_internal"."mz_frontiers"]␠("object_id") +mz_hydration_statuses_ind CREATE␠INDEX␠"mz_hydration_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s765␠AS␠"mz_internal"."mz_hydration_statuses"]␠("object_id",␠"replica_id") +mz_indexes_ind CREATE␠INDEX␠"mz_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s489␠AS␠"mz_catalog"."mz_indexes"]␠("id") +mz_kafka_sources_ind CREATE␠INDEX␠"mz_kafka_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s483␠AS␠"mz_catalog"."mz_kafka_sources"]␠("id") +mz_materialized_views_ind CREATE␠INDEX␠"mz_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s547␠AS␠"mz_catalog"."mz_materialized_views"]␠("id") mz_message_batch_counts_received_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_batch_counts_received_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_batch_counts_received_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id") mz_message_batch_counts_sent_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_batch_counts_sent_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_batch_counts_sent_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id") mz_message_counts_received_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_counts_received_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_counts_received_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id") mz_message_counts_sent_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_counts_sent_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_counts_sent_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id") -mz_notices_ind CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s839␠AS␠"mz_internal"."mz_notices"]␠("id") -mz_object_arrangement_size_history_object_ind CREATE␠INDEX␠"mz_object_arrangement_size_history_object_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s756␠AS␠"mz_internal"."mz_object_arrangement_size_history"]␠("object_id") -mz_object_arrangement_size_history_ts_ind CREATE␠INDEX␠"mz_object_arrangement_size_history_ts_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s756␠AS␠"mz_internal"."mz_object_arrangement_size_history"]␠("collection_timestamp") -mz_object_arrangement_sizes_ind CREATE␠INDEX␠"mz_object_arrangement_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s754␠AS␠"mz_internal"."mz_object_arrangement_sizes"]␠("replica_id") -mz_object_dependencies_ind CREATE␠INDEX␠"mz_object_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s483␠AS␠"mz_internal"."mz_object_dependencies"]␠("object_id") -mz_object_graph_edges_ind CREATE␠INDEX␠"mz_object_graph_edges_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s745␠AS␠"mz_internal"."mz_object_graph_edges"]␠("object_id") -mz_object_history_ind CREATE␠INDEX␠"mz_object_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s557␠AS␠"mz_internal"."mz_object_history"]␠("id") -mz_object_lifetimes_ind CREATE␠INDEX␠"mz_object_lifetimes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s558␠AS␠"mz_internal"."mz_object_lifetimes"]␠("id") -mz_object_transitive_dependencies_ind CREATE␠INDEX␠"mz_object_transitive_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s574␠AS␠"mz_internal"."mz_object_transitive_dependencies"]␠("object_id") -mz_objects_ind CREATE␠INDEX␠"mz_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s554␠AS␠"mz_catalog"."mz_objects"]␠("schema_id") +mz_notices_ind CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s840␠AS␠"mz_internal"."mz_notices"]␠("id") +mz_object_arrangement_size_history_object_ind CREATE␠INDEX␠"mz_object_arrangement_size_history_object_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s757␠AS␠"mz_internal"."mz_object_arrangement_size_history"]␠("object_id") +mz_object_arrangement_size_history_ts_ind CREATE␠INDEX␠"mz_object_arrangement_size_history_ts_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s757␠AS␠"mz_internal"."mz_object_arrangement_size_history"]␠("collection_timestamp") +mz_object_arrangement_sizes_ind CREATE␠INDEX␠"mz_object_arrangement_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s755␠AS␠"mz_internal"."mz_object_arrangement_sizes"]␠("replica_id") +mz_object_dependencies_ind CREATE␠INDEX␠"mz_object_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s484␠AS␠"mz_internal"."mz_object_dependencies"]␠("object_id") +mz_object_graph_edges_ind CREATE␠INDEX␠"mz_object_graph_edges_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s746␠AS␠"mz_internal"."mz_object_graph_edges"]␠("object_id") +mz_object_history_ind CREATE␠INDEX␠"mz_object_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s558␠AS␠"mz_internal"."mz_object_history"]␠("id") +mz_object_lifetimes_ind CREATE␠INDEX␠"mz_object_lifetimes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s559␠AS␠"mz_internal"."mz_object_lifetimes"]␠("id") +mz_object_transitive_dependencies_ind CREATE␠INDEX␠"mz_object_transitive_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s575␠AS␠"mz_internal"."mz_object_transitive_dependencies"]␠("object_id") +mz_objects_ind CREATE␠INDEX␠"mz_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s555␠AS␠"mz_catalog"."mz_objects"]␠("schema_id") mz_peek_durations_histogram_raw_s2_primary_idx CREATE␠INDEX␠"mz_peek_durations_histogram_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_peek_durations_histogram_raw"␠("worker_id",␠"type",␠"duration_ns") -mz_recent_activity_log_thinned_ind CREATE␠INDEX␠"mz_recent_activity_log_thinned_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s718␠AS␠"mz_internal"."mz_recent_activity_log_thinned"]␠("sql_hash") -mz_recent_sql_text_ind CREATE␠INDEX␠"mz_recent_sql_text_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s714␠AS␠"mz_internal"."mz_recent_sql_text"]␠("sql_hash") -mz_recent_storage_usage_ind CREATE␠INDEX␠"mz_recent_storage_usage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s832␠AS␠"mz_catalog"."mz_recent_storage_usage"]␠("object_id") -mz_roles_ind CREATE␠INDEX␠"mz_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s506␠AS␠"mz_catalog"."mz_roles"]␠("id") +mz_recent_activity_log_thinned_ind CREATE␠INDEX␠"mz_recent_activity_log_thinned_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s719␠AS␠"mz_internal"."mz_recent_activity_log_thinned"]␠("sql_hash") +mz_recent_sql_text_ind CREATE␠INDEX␠"mz_recent_sql_text_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s715␠AS␠"mz_internal"."mz_recent_sql_text"]␠("sql_hash") +mz_recent_storage_usage_ind CREATE␠INDEX␠"mz_recent_storage_usage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s833␠AS␠"mz_catalog"."mz_recent_storage_usage"]␠("object_id") +mz_roles_ind CREATE␠INDEX␠"mz_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s507␠AS␠"mz_catalog"."mz_roles"]␠("id") mz_scheduling_elapsed_raw_s2_primary_idx CREATE␠INDEX␠"mz_scheduling_elapsed_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_scheduling_elapsed_raw"␠("id",␠"worker_id") mz_scheduling_parks_histogram_raw_s2_primary_idx CREATE␠INDEX␠"mz_scheduling_parks_histogram_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_scheduling_parks_histogram_raw"␠("worker_id",␠"slept_for_ns",␠"requested_ns") -mz_schemas_ind CREATE␠INDEX␠"mz_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s486␠AS␠"mz_catalog"."mz_schemas"]␠("database_id") -mz_secrets_ind CREATE␠INDEX␠"mz_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s521␠AS␠"mz_catalog"."mz_secrets"]␠("name") -mz_show_all_objects_ind CREATE␠INDEX␠"mz_show_all_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s602␠AS␠"mz_internal"."mz_show_all_objects"]␠("schema_id") -mz_show_cluster_replicas_ind CREATE␠INDEX␠"mz_show_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s766␠AS␠"mz_internal"."mz_show_cluster_replicas"]␠("cluster") -mz_show_clusters_ind CREATE␠INDEX␠"mz_show_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s604␠AS␠"mz_internal"."mz_show_clusters"]␠("name") -mz_show_columns_ind CREATE␠INDEX␠"mz_show_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s603␠AS␠"mz_internal"."mz_show_columns"]␠("id") -mz_show_connections_ind CREATE␠INDEX␠"mz_show_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s612␠AS␠"mz_internal"."mz_show_connections"]␠("schema_id") -mz_show_databases_ind CREATE␠INDEX␠"mz_show_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s606␠AS␠"mz_internal"."mz_show_databases"]␠("name") -mz_show_indexes_ind CREATE␠INDEX␠"mz_show_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s616␠AS␠"mz_internal"."mz_show_indexes"]␠("schema_id") -mz_show_materialized_views_ind CREATE␠INDEX␠"mz_show_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s615␠AS␠"mz_internal"."mz_show_materialized_views"]␠("schema_id") -mz_show_roles_ind CREATE␠INDEX␠"mz_show_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s611␠AS␠"mz_internal"."mz_show_roles"]␠("name") -mz_show_schemas_ind CREATE␠INDEX␠"mz_show_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s607␠AS␠"mz_internal"."mz_show_schemas"]␠("database_id") -mz_show_secrets_ind CREATE␠INDEX␠"mz_show_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s605␠AS␠"mz_internal"."mz_show_secrets"]␠("schema_id") -mz_show_sinks_ind CREATE␠INDEX␠"mz_show_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s614␠AS␠"mz_internal"."mz_show_sinks"]␠("schema_id") -mz_show_sources_ind CREATE␠INDEX␠"mz_show_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s613␠AS␠"mz_internal"."mz_show_sources"]␠("schema_id") -mz_show_tables_ind CREATE␠INDEX␠"mz_show_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s608␠AS␠"mz_internal"."mz_show_tables"]␠("schema_id") -mz_show_types_ind CREATE␠INDEX␠"mz_show_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s610␠AS␠"mz_internal"."mz_show_types"]␠("schema_id") -mz_show_views_ind CREATE␠INDEX␠"mz_show_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s609␠AS␠"mz_internal"."mz_show_views"]␠("schema_id") -mz_sink_statistics_ind CREATE␠INDEX␠"mz_sink_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s731␠AS␠"mz_internal"."mz_sink_statistics"]␠("id",␠"replica_id") -mz_sink_status_history_ind CREATE␠INDEX␠"mz_sink_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s703␠AS␠"mz_internal"."mz_sink_status_history"]␠("sink_id") -mz_sink_statuses_ind CREATE␠INDEX␠"mz_sink_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s704␠AS␠"mz_internal"."mz_sink_statuses"]␠("id") -mz_sinks_ind CREATE␠INDEX␠"mz_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s498␠AS␠"mz_catalog"."mz_sinks"]␠("id") -mz_source_statistics_ind CREATE␠INDEX␠"mz_source_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s729␠AS␠"mz_internal"."mz_source_statistics"]␠("id",␠"replica_id") -mz_source_statistics_with_history_ind CREATE␠INDEX␠"mz_source_statistics_with_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s727␠AS␠"mz_internal"."mz_source_statistics_with_history"]␠("id",␠"replica_id") -mz_source_status_history_ind CREATE␠INDEX␠"mz_source_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s705␠AS␠"mz_internal"."mz_source_status_history"]␠("source_id") -mz_source_statuses_ind CREATE␠INDEX␠"mz_source_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s722␠AS␠"mz_internal"."mz_source_statuses"]␠("id") -mz_sources_ind CREATE␠INDEX␠"mz_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s491␠AS␠"mz_catalog"."mz_sources"]␠("id") -mz_tables_ind CREATE␠INDEX␠"mz_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s490␠AS␠"mz_catalog"."mz_tables"]␠("schema_id") -mz_types_ind CREATE␠INDEX␠"mz_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s500␠AS␠"mz_catalog"."mz_types"]␠("schema_id") -mz_views_ind CREATE␠INDEX␠"mz_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s499␠AS␠"mz_catalog"."mz_views"]␠("schema_id") -mz_wallclock_global_lag_recent_history_ind CREATE␠INDEX␠"mz_wallclock_global_lag_recent_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s738␠AS␠"mz_internal"."mz_wallclock_global_lag_recent_history"]␠("object_id") -mz_webhook_sources_ind CREATE␠INDEX␠"mz_webhook_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s544␠AS␠"mz_internal"."mz_webhook_sources"]␠("id") -pg_attrdef_all_databases_ind CREATE␠INDEX␠"pg_attrdef_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s643␠AS␠"mz_internal"."pg_attrdef_all_databases"]␠("oid",␠"adrelid",␠"adnum",␠"adbin",␠"adsrc") -pg_attribute_all_databases_ind CREATE␠INDEX␠"pg_attribute_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s636␠AS␠"mz_internal"."pg_attribute_all_databases"]␠("attrelid",␠"attname",␠"atttypid",␠"attlen",␠"attnum",␠"atttypmod",␠"attnotnull",␠"atthasdef",␠"attidentity",␠"attgenerated",␠"attisdropped",␠"attcollation",␠"database_name",␠"pg_type_database_name") -pg_authid_core_ind CREATE␠INDEX␠"pg_authid_core_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s653␠AS␠"mz_internal"."pg_authid_core"]␠("rolname") -pg_class_all_databases_ind CREATE␠INDEX␠"pg_class_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s624␠AS␠"mz_internal"."pg_class_all_databases"]␠("relname") -pg_description_all_databases_ind CREATE␠INDEX␠"pg_description_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s633␠AS␠"mz_internal"."pg_description_all_databases"]␠("objoid",␠"classoid",␠"objsubid",␠"description",␠"oid_database_name",␠"class_database_name") -pg_namespace_all_databases_ind CREATE␠INDEX␠"pg_namespace_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s621␠AS␠"mz_internal"."pg_namespace_all_databases"]␠("nspname") -pg_type_all_databases_ind CREATE␠INDEX␠"pg_type_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s630␠AS␠"mz_internal"."pg_type_all_databases"]␠("oid") +mz_schemas_ind CREATE␠INDEX␠"mz_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s487␠AS␠"mz_catalog"."mz_schemas"]␠("database_id") +mz_secrets_ind CREATE␠INDEX␠"mz_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s522␠AS␠"mz_catalog"."mz_secrets"]␠("name") +mz_show_all_objects_ind CREATE␠INDEX␠"mz_show_all_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s603␠AS␠"mz_internal"."mz_show_all_objects"]␠("schema_id") +mz_show_cluster_replicas_ind CREATE␠INDEX␠"mz_show_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s767␠AS␠"mz_internal"."mz_show_cluster_replicas"]␠("cluster") +mz_show_clusters_ind CREATE␠INDEX␠"mz_show_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s605␠AS␠"mz_internal"."mz_show_clusters"]␠("name") +mz_show_columns_ind CREATE␠INDEX␠"mz_show_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s604␠AS␠"mz_internal"."mz_show_columns"]␠("id") +mz_show_connections_ind CREATE␠INDEX␠"mz_show_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s613␠AS␠"mz_internal"."mz_show_connections"]␠("schema_id") +mz_show_databases_ind CREATE␠INDEX␠"mz_show_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s607␠AS␠"mz_internal"."mz_show_databases"]␠("name") +mz_show_indexes_ind CREATE␠INDEX␠"mz_show_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s617␠AS␠"mz_internal"."mz_show_indexes"]␠("schema_id") +mz_show_materialized_views_ind CREATE␠INDEX␠"mz_show_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s616␠AS␠"mz_internal"."mz_show_materialized_views"]␠("schema_id") +mz_show_roles_ind CREATE␠INDEX␠"mz_show_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s612␠AS␠"mz_internal"."mz_show_roles"]␠("name") +mz_show_schemas_ind CREATE␠INDEX␠"mz_show_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s608␠AS␠"mz_internal"."mz_show_schemas"]␠("database_id") +mz_show_secrets_ind CREATE␠INDEX␠"mz_show_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s606␠AS␠"mz_internal"."mz_show_secrets"]␠("schema_id") +mz_show_sinks_ind CREATE␠INDEX␠"mz_show_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s615␠AS␠"mz_internal"."mz_show_sinks"]␠("schema_id") +mz_show_sources_ind CREATE␠INDEX␠"mz_show_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s614␠AS␠"mz_internal"."mz_show_sources"]␠("schema_id") +mz_show_tables_ind CREATE␠INDEX␠"mz_show_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s609␠AS␠"mz_internal"."mz_show_tables"]␠("schema_id") +mz_show_types_ind CREATE␠INDEX␠"mz_show_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s611␠AS␠"mz_internal"."mz_show_types"]␠("schema_id") +mz_show_views_ind CREATE␠INDEX␠"mz_show_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s610␠AS␠"mz_internal"."mz_show_views"]␠("schema_id") +mz_sink_statistics_ind CREATE␠INDEX␠"mz_sink_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s732␠AS␠"mz_internal"."mz_sink_statistics"]␠("id",␠"replica_id") +mz_sink_status_history_ind CREATE␠INDEX␠"mz_sink_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s704␠AS␠"mz_internal"."mz_sink_status_history"]␠("sink_id") +mz_sink_statuses_ind CREATE␠INDEX␠"mz_sink_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s705␠AS␠"mz_internal"."mz_sink_statuses"]␠("id") +mz_sinks_ind CREATE␠INDEX␠"mz_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s499␠AS␠"mz_catalog"."mz_sinks"]␠("id") +mz_source_statistics_ind CREATE␠INDEX␠"mz_source_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s730␠AS␠"mz_internal"."mz_source_statistics"]␠("id",␠"replica_id") +mz_source_statistics_with_history_ind CREATE␠INDEX␠"mz_source_statistics_with_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s728␠AS␠"mz_internal"."mz_source_statistics_with_history"]␠("id",␠"replica_id") +mz_source_status_history_ind CREATE␠INDEX␠"mz_source_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s706␠AS␠"mz_internal"."mz_source_status_history"]␠("source_id") +mz_source_statuses_ind CREATE␠INDEX␠"mz_source_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s723␠AS␠"mz_internal"."mz_source_statuses"]␠("id") +mz_sources_ind CREATE␠INDEX␠"mz_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s492␠AS␠"mz_catalog"."mz_sources"]␠("id") +mz_tables_ind CREATE␠INDEX␠"mz_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s491␠AS␠"mz_catalog"."mz_tables"]␠("schema_id") +mz_types_ind CREATE␠INDEX␠"mz_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s501␠AS␠"mz_catalog"."mz_types"]␠("schema_id") +mz_views_ind CREATE␠INDEX␠"mz_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s500␠AS␠"mz_catalog"."mz_views"]␠("schema_id") +mz_wallclock_global_lag_recent_history_ind CREATE␠INDEX␠"mz_wallclock_global_lag_recent_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s739␠AS␠"mz_internal"."mz_wallclock_global_lag_recent_history"]␠("object_id") +mz_webhook_sources_ind CREATE␠INDEX␠"mz_webhook_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s545␠AS␠"mz_internal"."mz_webhook_sources"]␠("id") +pg_attrdef_all_databases_ind CREATE␠INDEX␠"pg_attrdef_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s644␠AS␠"mz_internal"."pg_attrdef_all_databases"]␠("oid",␠"adrelid",␠"adnum",␠"adbin",␠"adsrc") +pg_attribute_all_databases_ind CREATE␠INDEX␠"pg_attribute_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s637␠AS␠"mz_internal"."pg_attribute_all_databases"]␠("attrelid",␠"attname",␠"atttypid",␠"attlen",␠"attnum",␠"atttypmod",␠"attnotnull",␠"atthasdef",␠"attidentity",␠"attgenerated",␠"attisdropped",␠"attcollation",␠"database_name",␠"pg_type_database_name") +pg_authid_core_ind CREATE␠INDEX␠"pg_authid_core_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s654␠AS␠"mz_internal"."pg_authid_core"]␠("rolname") +pg_class_all_databases_ind CREATE␠INDEX␠"pg_class_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s625␠AS␠"mz_internal"."pg_class_all_databases"]␠("relname") +pg_description_all_databases_ind CREATE␠INDEX␠"pg_description_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s634␠AS␠"mz_internal"."pg_description_all_databases"]␠("objoid",␠"classoid",␠"objsubid",␠"description",␠"oid_database_name",␠"class_database_name") +pg_namespace_all_databases_ind CREATE␠INDEX␠"pg_namespace_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s622␠AS␠"mz_internal"."pg_namespace_all_databases"]␠("nspname") +pg_type_all_databases_ind CREATE␠INDEX␠"pg_type_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s631␠AS␠"mz_internal"."pg_type_all_databases"]␠("oid") # Record all transitive dependencies (tables, sources, views, mvs) of indexes on # the mz_catalog_server cluster. @@ -365,6 +366,12 @@ mz_compute_import_frontiers_per_worker export_id mz_compute_import_frontiers_per_worker import_id mz_compute_import_frontiers_per_worker time mz_compute_import_frontiers_per_worker worker_id +mz_compute_lifecycle_events_per_worker details +mz_compute_lifecycle_events_per_worker event +mz_compute_lifecycle_events_per_worker export_id +mz_compute_lifecycle_events_per_worker occurred_at +mz_compute_lifecycle_events_per_worker reason +mz_compute_lifecycle_events_per_worker worker_id mz_compute_lir_mapping_per_worker global_id mz_compute_lir_mapping_per_worker lir_id mz_compute_lir_mapping_per_worker nesting diff --git a/test/sqllogictest/oid.slt b/test/sqllogictest/oid.slt index 1d99040fdb6d9..d1933ef36c0ea 100644 --- a/test/sqllogictest/oid.slt +++ b/test/sqllogictest/oid.slt @@ -1249,3 +1249,4 @@ SELECT oid, name FROM mz_objects WHERE id LIKE 's%' AND oid < 20000 ORDER BY oid 17117 mz_object_graph_edges_ind 17118 mz_builtin_tables 17119 mz_builtin_views +17120 mz_compute_lifecycle_events_per_worker diff --git a/test/sqllogictest/pg_catalog_user.slt b/test/sqllogictest/pg_catalog_user.slt index d892d2d48689a..b09c3b5f32729 100644 --- a/test/sqllogictest/pg_catalog_user.slt +++ b/test/sqllogictest/pg_catalog_user.slt @@ -27,7 +27,7 @@ CREATE ROLE "materialize@foocorp.io" WITH LOGIN query TIBBBBTTT rowsort SELECT usename, usesysid, usecreatedb, usesuper, userepl, usebypassrls, passwd, valuntil, useconfig FROM pg_user; ---- -materialize@foocorp.io 20196 false NULL false false ******** NULL NULL +materialize@foocorp.io 20202 false NULL false false ******** NULL NULL mz_support 16662 false true false false ******** NULL NULL mz_system 16661 true true false false ******** NULL NULL diff --git a/test/sqllogictest/regclass.slt b/test/sqllogictest/regclass.slt index d3816146ae8ea..7257fab6c0108 100644 --- a/test/sqllogictest/regclass.slt +++ b/test/sqllogictest/regclass.slt @@ -35,12 +35,12 @@ CREATE MATERIALIZED VIEW s.m AS SELECT * FROM s.t; query T SELECT 't'::regclass::oid::int ---- -20196 +20202 query T SELECT 's.t'::regclass::oid::int ---- -20197 +20203 query T SELECT 't'::regclass = 's.t'::regclass @@ -73,7 +73,7 @@ t query T SELECT 't'::regclass::oid::int ---- -20197 +20203 query T SELECT 'public.t'::regclass::text; @@ -101,12 +101,12 @@ d.public.t query T SELECT 'm'::regclass::oid::int ---- -20201 +20207 query T SELECT 's.m'::regclass::oid::int ---- -20202 +20208 query T SELECT 'm'::regclass = 's.m'::regclass @@ -296,7 +296,7 @@ true query T SELECT 'materialize.public.t'::regclass::oid::int ---- -20196 +20202 query error relation "t" does not exist SELECT 't'::regclass::oid::int diff --git a/test/sqllogictest/regtype.slt b/test/sqllogictest/regtype.slt index 8ba6b5b88e8f3..98aa4a37e3671 100644 --- a/test/sqllogictest/regtype.slt +++ b/test/sqllogictest/regtype.slt @@ -142,12 +142,12 @@ CREATE TYPE d.public.t AS LIST (ELEMENT TYPE = int4); query T SELECT 't'::regtype::oid::int ---- -20197 +20203 query T SELECT 's.t'::regtype::oid::int ---- -20198 +20204 query T SELECT 't'::regtype = 's.t'::regtype diff --git a/test/testdrive/catalog.td b/test/testdrive/catalog.td index 1ecf361507c03..8828457f52c61 100644 --- a/test/testdrive/catalog.td +++ b/test/testdrive/catalog.td @@ -758,6 +758,7 @@ mz_compute_exports_per_worker log "" mz_compute_frontiers_per_worker log "" mz_compute_hydration_times_per_worker log "" mz_compute_import_frontiers_per_worker log "" +mz_compute_lifecycle_events_per_worker log "" mz_compute_lir_mapping_per_worker log "" mz_compute_operator_durations_histogram_raw log "" mz_compute_operator_hydration_statuses_per_worker log "" @@ -837,7 +838,7 @@ test_table "" # There is one entry in mz_indexes for each field_number/expression of the index. > SELECT COUNT(id) FROM mz_indexes WHERE id LIKE 's%' -273 +279 # Create a second schema with the same table name as above > CREATE SCHEMA tester2 diff --git a/test/testdrive/compute-lifecycle-events.td b/test/testdrive/compute-lifecycle-events.td new file mode 100644 index 0000000000000..a22ab27ca8368 --- /dev/null +++ b/test/testdrive/compute-lifecycle-events.td @@ -0,0 +1,168 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# Test the lifecycle event log reported by +# `mz_introspection.mz_compute_lifecycle_events_per_worker`. +# +# These tests rely on testdrive's retry feature, as dataflows take an unknown +# (but hopefully small) time to be installed, to hydrate, and to write. The +# sections that must not retry, so that a transient invariant violation is not +# retried away, come last, since `set-max-tries` has no way to restore the +# default. + +$ set-sql-timeout duration=60s + +> CREATE CLUSTER test SIZE 'scale=1,workers=2' +> SET cluster = test + +> CREATE TABLE t (a int) + +# An index has no persist sink, so its lifecycle stops at `hydrated`. Every +# worker hydrates its own fragment of the dataflow, so each of the three stages +# is reported once per worker. + +> CREATE INDEX idx IN CLUSTER test ON t (a) + +> SELECT l.event, count(*) + FROM mz_introspection.mz_compute_lifecycle_events_per_worker l + JOIN mz_indexes i ON (i.id = l.export_id) + WHERE i.name = 'idx' + GROUP BY l.event +installed 2 +started 2 +hydrated 2 + +# The stages are ordered within each worker, and `occurred_at` is a wallclock +# instant rather than an offset from some arbitrary origin. + +> SELECT DISTINCT + max(l.occurred_at) FILTER (WHERE l.event = 'started') + >= max(l.occurred_at) FILTER (WHERE l.event = 'installed'), + max(l.occurred_at) FILTER (WHERE l.event = 'hydrated') + >= max(l.occurred_at) FILTER (WHERE l.event = 'started'), + max(l.occurred_at) BETWEEN now() - '1 hour'::interval AND now() + FROM mz_introspection.mz_compute_lifecycle_events_per_worker l + JOIN mz_indexes i ON (i.id = l.export_id) + WHERE i.name = 'idx' + GROUP BY l.worker_id +true true true + +# Every event carries the dataflow's as-of, without which the interval between +# two stages says nothing about how much work was done. + +> SELECT DISTINCT l.details->>'as_of' IS NOT NULL + FROM mz_introspection.mz_compute_lifecycle_events_per_worker l + JOIN mz_indexes i ON (i.id = l.export_id) + WHERE i.name = 'idx' +true + +# A materialized view writes, so it reaches `written`. The write stages are +# observed by the single worker that maintains the sink frontier, so they are +# reported once per object rather than once per worker. + +> CREATE MATERIALIZED VIEW mv IN CLUSTER test AS SELECT a + 1 AS a FROM t + +> SELECT l.event, count(*) + FROM mz_introspection.mz_compute_lifecycle_events_per_worker l + JOIN mz_materialized_views mv ON (mv.id = l.export_id) + WHERE mv.name = 'mv' AND l.event = 'written' + GROUP BY l.event +written 1 + +> SELECT l.event, count(*) + FROM mz_introspection.mz_compute_lifecycle_events_per_worker l + JOIN mz_materialized_views mv ON (mv.id = l.export_id) + WHERE mv.name = 'mv' AND l.event = 'hydrated' + GROUP BY l.event +hydrated 2 + +# `written` never precedes `hydrated`, which is what makes the interval between +# them the time taken to make the hydrated output durable. +# +# The comparison has to be per worker. `written` is reported only by the worker +# that maintains the sink frontier, and it is ordered after that worker's own +# `hydrated`. Another worker may hydrate later still, so comparing `written` +# against the maximum `hydrated` across all workers proves nothing. + +> SELECT DISTINCT + max(l.occurred_at) FILTER (WHERE l.event = 'written') + >= max(l.occurred_at) FILTER (WHERE l.event = 'hydrated') + FROM mz_introspection.mz_compute_lifecycle_events_per_worker l + JOIN mz_materialized_views mv ON (mv.id = l.export_id) + WHERE mv.name = 'mv' + GROUP BY l.worker_id + HAVING count(*) FILTER (WHERE l.event = 'written') > 0 +true + +# `write_blocked` is only reported for an object that has hydrated and is still +# not permitted to write, which is a replica in read-only mode awaiting a +# cutover. Testdrive cannot reach that state, so there is nothing to assert +# positively here. The invariant block at the end of this file requires the +# blocked/unblocked pair to be absent or complete, never that it is present. + +# Dropping an object retracts its rows, so the log does not accumulate +# lifecycles of objects that no longer exist. Transient dataflows are excluded: +# a peek installs one with a `t` prefixed export id that is never in +# `mz_objects`, so it would read as an orphan for as long as it lives. + +> DROP MATERIALIZED VIEW mv +> DROP INDEX idx + +> SELECT count(*) + FROM mz_introspection.mz_compute_lifecycle_events_per_worker l + LEFT JOIN mz_objects o ON (o.id = l.export_id) + WHERE o.id IS NULL AND l.export_id NOT LIKE 't%' +0 + +# Invariants that must hold at all times. Retries are disabled from here on, so +# that a violation cannot be retried away. + +$ set-max-tries max-tries=1 + +# `reason` is drawn from a closed vocabulary, and only the events that have a +# cause to report carry one. + +> SELECT count(*) + FROM mz_introspection.mz_compute_lifecycle_events_per_worker + WHERE event NOT IN ( + 'installed', 'started', 'hydrated', + 'write_blocked', 'write_unblocked', 'written' + ) + OR reason NOT IN ('read_only') + OR (reason IS NOT NULL AND event <> 'write_blocked') +0 + +# No worker reports a stage without its predecessors. + +> SELECT count(*) + FROM ( + SELECT + export_id, + worker_id, + array_agg(event) AS events + FROM mz_introspection.mz_compute_lifecycle_events_per_worker + GROUP BY export_id, worker_id + ) + WHERE NOT ('installed' = ANY(events)) + OR ('hydrated' = ANY(events) AND NOT 'started' = ANY(events)) + OR ('write_unblocked' = ANY(events) AND NOT 'write_blocked' = ANY(events)) + OR ('written' = ANY(events) AND NOT 'hydrated' = ANY(events)) +0 + +# A stage is reported at most once per export and worker, which is what lets a +# reader take an event's `occurred_at` without aggregating first. + +> SELECT count(*) + FROM ( + SELECT export_id, worker_id, event, count(*) AS n + FROM mz_introspection.mz_compute_lifecycle_events_per_worker + GROUP BY export_id, worker_id, event + ) + WHERE n > 1 +0 diff --git a/test/testdrive/indexes.td b/test/testdrive/indexes.td index d54a1752031be..57f3c9897ea64 100644 --- a/test/testdrive/indexes.td +++ b/test/testdrive/indexes.td @@ -319,6 +319,7 @@ mz_compute_frontiers_per_worker_s2_primary_idx mz_compute_frontiers mz_compute_hydration_times_ind mz_compute_hydration_times mz_catalog_server {replica_id} "" mz_compute_hydration_times_per_worker_s2_primary_idx mz_compute_hydration_times_per_worker mz_catalog_server {export_id,worker_id} "" mz_compute_import_frontiers_per_worker_s2_primary_idx mz_compute_import_frontiers_per_worker mz_catalog_server {export_id,import_id,worker_id} "" +mz_compute_lifecycle_events_per_worker_s2_primary_idx mz_compute_lifecycle_events_per_worker mz_catalog_server {export_id,worker_id,event,occurred_at,reason,details} "" mz_compute_lir_mapping_per_worker_s2_primary_idx mz_compute_lir_mapping_per_worker mz_catalog_server {global_id,lir_id,worker_id} "" mz_compute_operator_durations_histogram_raw_s2_primary_idx mz_compute_operator_durations_histogram_raw mz_catalog_server {id,worker_id,duration_ns} "" mz_compute_operator_hydration_statuses_per_worker_s2_primary_idx mz_compute_operator_hydration_statuses_per_worker mz_catalog_server {export_id,lir_id,worker_id} "" diff --git a/test/workload-replay/objects.txt b/test/workload-replay/objects.txt index 8bdd9443e0bd7..5f5130b6b7760 100644 --- a/test/workload-replay/objects.txt +++ b/test/workload-replay/objects.txt @@ -451,6 +451,13 @@ mz_compute_import_frontiers_per_worker_s3_primary_idx mz_compute_import_frontiers_per_worker_s4_primary_idx mz_compute_import_frontiers_per_worker_s5_primary_idx mz_compute_import_frontiers_per_worker_u1_primary_idx +mz_compute_lifecycle_events_per_worker +mz_compute_lifecycle_events_per_worker_s1_primary_idx +mz_compute_lifecycle_events_per_worker_s2_primary_idx +mz_compute_lifecycle_events_per_worker_s3_primary_idx +mz_compute_lifecycle_events_per_worker_s4_primary_idx +mz_compute_lifecycle_events_per_worker_s5_primary_idx +mz_compute_lifecycle_events_per_worker_u1_primary_idx mz_compute_lir_mapping_per_worker mz_compute_lir_mapping_per_worker_s1_primary_idx mz_compute_lir_mapping_per_worker_s2_primary_idx diff --git a/test/workload-replay/system_catalog_identifiers.txt b/test/workload-replay/system_catalog_identifiers.txt index 730a57364f729..4afc64da9ee79 100644 --- a/test/workload-replay/system_catalog_identifiers.txt +++ b/test/workload-replay/system_catalog_identifiers.txt @@ -755,6 +755,13 @@ mz_compute_import_frontiers_per_worker_s3_primary_idx mz_compute_import_frontiers_per_worker_s4_primary_idx mz_compute_import_frontiers_per_worker_s5_primary_idx mz_compute_import_frontiers_per_worker_u1_primary_idx +mz_compute_lifecycle_events_per_worker +mz_compute_lifecycle_events_per_worker_s1_primary_idx +mz_compute_lifecycle_events_per_worker_s2_primary_idx +mz_compute_lifecycle_events_per_worker_s3_primary_idx +mz_compute_lifecycle_events_per_worker_s4_primary_idx +mz_compute_lifecycle_events_per_worker_s5_primary_idx +mz_compute_lifecycle_events_per_worker_u1_primary_idx mz_compute_lir_mapping_per_worker mz_compute_lir_mapping_per_worker_s1_primary_idx mz_compute_lir_mapping_per_worker_s2_primary_idx