diff --git a/doc/developer/design/20260817_compute_hydration_timestamps.md b/doc/developer/design/20260817_compute_hydration_timestamps.md index 0ac81a5990ac5..c4ca909d8e45e 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. First, some stages are per worker and others are not. +Each worker computes its own fragment of the dataflow, so `installed`, `started` and +`snapshot_complete` happen once per worker, while whether the output is durable is a +property of the sink as a whole. Second, 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,134 @@ 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 +dataflow_id uint8 not null +event text not null +occurred_at timestamptz not null +reason text nullable +details jsonb nullable +``` + +| event | reported | `reason` | +| --- | --- | --- | +| `installed` | per worker | none | +| `started` | per worker | none | +| `snapshot_complete` | 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. 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. + +**`worker_id`.** Every event carries the worker that recorded it, which is not the same as +the event being specific to that worker. How to read it depends on the event: a per-worker +stage describes that worker's fragment of the dataflow, while a per-object stage is +recorded by the one worker that maintains the sink's write frontier and describes the +export as a whole. + +**`dataflow_id`.** `snapshot_complete` and the write stages are properties of one export. +`installed` and `started` describe the dataflow, so they carry one instant shared by every +export it maintains. Today a non-logging dataflow has at most one export, so the +distinction only starts to matter if that changes. + +The relation is keyed by export, since the export id is what `mz_objects.id` holds, and +carries `dataflow_id` as a column so that the shared events are recognizable as shared: +`SELECT DISTINCT dataflow_id, event, occurred_at` recovers the dataflow-level facts. + +**`installed` is the denominator.** Every worker logs `installed` when the export is +created, so the count of `installed` events for an export is the number of workers +reporting on it: + +```sql +count(*) FILTER (WHERE event = 'snapshot_complete') + = count(*) FILTER (WHERE event = 'installed') +``` + +is the all-workers-reported test, with no worker count needed from the catalog. The +per-object stages are expected exactly once, or not at all for an export that never +writes. Which stages are reported per worker is a fixed property of the vocabulary, so a +consumer encodes it once rather than deriving it per query. + +**Which frontier each stage reads.** `snapshot_complete` reads the dataflow's own progress +frontier, the compute probe, not the reported output frontier. The output frontier is the +meet of the write and compute frontiers, which makes it a measure of durability rather +than of computation. 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 +`snapshot_complete` coincides with durability. + +`written` reads the output shard's upper passing the as-of. It says the output is durable +through the as-of, not that this replica wrote it: every replica's `mint` reads the same +upper back from persist, so it advances on all of them when any one wins the append. + +**The write stages are reported by the sink, and are ordered against nothing.** `mint` is +the only place that tracks the shard's upper, and it runs on one elected worker, which is +what makes these three events one report per object rather than one per worker. Reporting +them there rather than from the collection also means nothing outside the sink needs to +know which worker was elected. + +The price is that only the compute stages remain ordered. `written` reads the shard's +upper, and for a shard that already holds data the as-of is bounded one step below it, so +`written` is true from installation. `apply_refresh` advances the upper of a `REFRESH` +materialized view before its dataflow computes anything, for the same effect. And +`write_blocked` needs the dataflow's desired frontier to pass that upper, which is strictly +later than observing it, so a replica awaiting a cutover reports `written` *before* +`write_blocked`. A consumer must not assume any order among the six beyond the compute +stages. + +**`write_blocked` is logged on entry, not on exit,** because the state an operator debugs +is the one that has not ended, and that state carries no `write_unblocked` row. Entry +means the sink has a batch to mint and read-only mode forbids writing it, which is +exactly the condition `maybe_mint_batch_description` already evaluates. Reporting every +read-only observation instead would put the pair on essentially every materialized view, +since collections start read-only and the controller releases them. + +**`write_unblocked` is when writing became permitted,** not when the first write happened, +and it is reported only for a sink that was seen to wait. `mint` produces a batch +description as soon as the desired frontier passes the persist frontier, so a separate +stamp for the first write would carry no information. + +**`reason` and `details`.** `reason` is a typed cause for the event, and is NULL unless the +event has one. Its only value is `read_only`, on `write_blocked`. `details` is a nullable +`jsonb` object, following `mz_source_statuses` and `mz_sink_statuses`. Its only key is +`as_of`, the dataflow's as-of as a string, which every stage is defined relative to: +without it `snapshot_complete - started` cannot distinguish a fast computation from one +whose as-of was already recent. Both vocabularies are open, so tests assert on `event`, +`reason` and `occurred_at` and never on `details`. + +**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 are follow-up work. + +### 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 one of the reasons `written` is not ordered against +`snapshot_complete`. + ### A new hydration start event There is no event for hydration start today. Add @@ -274,10 +416,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 +458,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_snapshot` sees the dataflow's own progress frontier pass the as-of and logs the `snapshot_complete` stage | **`hydrated_at`**, and the `snapshot_complete` 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 +537,78 @@ 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 `snapshot_complete` 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". + +**Compatibility: what a consumer may rely on.** Downstream work builds rollups and +history relations on top of these two relations, so what is frozen and what may still +move needs saying explicitly rather than being inferred from the current +implementation. + +Stable. We intend these to hold, and will treat breaking one as a change that needs +coordinating with consumers rather than a detail: + +- The six `event` values, and their meanings. +- Which events are per worker and which are per object. `installed`, `started` and + `snapshot_complete` are per worker. `write_blocked`, `write_unblocked` and `written` + are per object, observed by the + elected frontier owner. This is a property of the event name, fixed for all + objects and all cluster shapes, so a consumer encodes it once rather than deriving + it per query. +- `installed` as the all-workers-reported denominator, per "`installed` is the + denominator" above. +- `occurred_at` is a wallclock instant, carrying its worker's epoch anchor. +- `snapshot_complete` is always the dataflow-progress reading and `written` is always + "the output is durable through the as-of". Neither varies by object type. +- The compute stages are ordered among themselves. Nothing else is ordered: not the two + sides against each other, and not the three write stages against each other. `written` + reads the output shard's upper, which moves whether or not this replica may write, so in + the one case that produces `write_blocked` at all, a replacement awaiting a cutover, the + shard already holds data and `written` is reported first. A consumer must not read the + six as one sequence, and must not treat a difference between two of them as an elapsed + interval unless it is between two compute stages. +- `mz_compute_hydration_times_per_worker.hydrated_at` and `time_ns` remain the + durability reading, computed in the demux. That relation will not become a view over + the lifecycle log. `mz_compute_hydration_statuses.hydrated` is `time_ns IS NOT NULL` + and the blue-green readiness query is defined on it, so pointing it at the earlier + dataflow-progress reading would have readiness cut over before the output is durable. +- `(export_id, worker_id)` is the exact join between the two relations. Both are per + worker, so joining on `export_id` alone multiplies rows by the worker count. +- Rows are retracted when the replica processes the drop, which is not when the + catalog transaction commits. `DROP` returns once the catalog row is gone, while the + retraction still has to reach the replica as an empty `AllowCompaction`, be logged by + the demux, and travel through the introspection subscribe and a storage append. A + consumer will therefore observe lifecycle rows whose `export_id` is no longer in + `mz_objects`, and must not use an inner join against the catalog to filter if losing + the tail of an episode matters. This is not specific to this relation: + `mz_compute_hydration_times_per_worker` rows also disappear only when the replica + retracts them. + +Open sets. A consumer must tolerate additions rather than enumerate these +exhaustively, even though the invariant tests assert closed vocabularies for the +values that exist today: + +- New `event` values. The two candidates are a per-worker durability stage, and a + stage recording that this replica's own append made the output durable. +- New `reason` values. Two useful attributions are not observable today and are + described under "Attributing why an export waited". +- New keys in `details`. + +The corollary for `written` is worth stating on its own, because it is the one place +where a follow-up could plausibly redefine an existing term. Attributing a write to +the replica that performed it, per "Attributing `written` to the replica that wrote", +arrives as a new stage. `written` keeps the meaning above. A consumer reading +`written` today will still be reading the same thing afterwards, and one that wants +attribution opts into the new stage. ### Why the replica and not the compute controller @@ -433,53 +636,37 @@ benefit of replica stamping is for same-version environmentd restarts, reconnections and generation changes, which is still the common case, rather than for upgrades. -### Implementation touch points - -- `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 - `handle_hydration_start`. Also a `CollectionLogging` method alongside - `set_hydrated`. -- `src/compute/src/compute_state.rs`: log the start event from `handle_schedule`, - from `handle_create_dataflow` when `import_ids` is empty, and from - `initialize_logging`. -- `src/compute-client/src/logging.rs`: the widened `RelationDesc`. - `LogVariant::desc` is the only exhaustive match a shape change touches, since - 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. - -Not touched, and deliberately so: the introspection subscribe, -`mz_internal.mz_compute_hydration_times`, -`mz_internal.mz_compute_hydration_statuses`, -`src/adapter/src/coord/message_handler.rs`, and -`src/mz-debug/src/system_catalog_dumper.rs`. - -New testdrive coverage worth adding: - -- `installed_at` set and `started_at` NULL for an object gated by - `HYDRATION_CONCURRENCY`. -- `started_at` NULL for an object waiting on an unavailable input. -- An import-free dataflow carrying `started_at` from creation, equal to its - `installed_at`, and satisfying the ordering invariant. -- Log collections having all three timestamps set. -- All timestamps surviving an environmentd restart unchanged. -- A replica restart yielding entirely fresh values. - -The most important tests are compatibility ones: that -`mz_compute_hydration_times_per_worker`, `mz_compute_hydration_times` and -`mz_compute_hydration_statuses` return identical values before and after. The -existing assertions in `test/testdrive/hydration-status.td` and the blue-green -tests are the contract and should be left alone rather than adjusted to fit. +### Goldens a new log relation touches + +Two of these are worth naming, because adding a *column* to an existing log reaches them +while leaving everything else alone, and neither is found by searching for a count. +`cluster.slt` lists each per-replica index's key columns with their positions, so a column +added to an unkeyed log shifts every position after it. `cockroach/srfs.slt` runs +`SELECT relname, unnest(indkey)`, so it gains a row per index instance. The reliable way to +find this class is to search every file that mentions the relation by name and read what it +asserts, rather than to reason about which kinds of value could have moved. ## Follow-up work +### Attributing `written` to the replica that wrote + +`written` says the output is durable and this replica was permitted to write, not that +this replica wrote, for the reason given above. Attribution needs a signal from the +replica's own successful append rather than a reading of the shared upper, and that +signal has to cross workers: `next_append_worker` rotates independently of +`sink::frontier_owner`, so the worker that appends is generally not the worker that +reports the stage. Whether the distinction is worth that machinery is the open +question, since the batches the replicas race to append are identical. + +### Attributing why an export waited + +Two causes the `reason` vocabulary would carry are not observable today. 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. Distinguishing a dataflow installed by a fresh `CreateDataflow` from one +retained across reconciliation is not observable in the replica at all, because a retained +dataflow emits no new `installed` event. + ### `mz_compute_hydration_timestamps`, the per-replica relation The per-worker relation is per replica and per worker, and lives in @@ -539,7 +726,7 @@ The prototype is the compute and catalog change itself, exercised through testdrive against a targeted replica. The validating query selects from `mz_compute_hydration_times_per_worker` on a cluster with a hydration concurrency limit and a handful of indexes, showing objects moving from waiting, -to hydrating, to hydrated, with the queueing interval visible separately from the +to computing, to snapshot complete, with the queueing interval visible separately from the hydration interval. A second run after an environmentd restart shows identical values, which is the property the design turns on. A third check confirms `mz_compute_hydration_times` and `mz_compute_hydration_statuses` are byte for byte @@ -591,38 +778,9 @@ inside every replica. cannot observe a transition, only its aftermath, so a hydration that starts and finishes between two samples is invisible. -## Settled during review - -Recorded so the reasoning is not relitigated. Each item is argued in the section -named. - -- **Stamping location.** The replica, not the compute controller. See "Why the - replica and not the compute controller". -- **`time_ns` is kept.** See "Preserving the existing relations". -- **`started_at` for dataflows with no imports is stamped at creation,** with the - backfill at hydration kept for the cases creation-time stamping cannot see. See - "A new hydration start event". -- **The per-worker log is widened in place,** keeping its name and OID, rather than - renamed behind a projecting view. See "Preserving the existing relations". -- **`hydration_time` is left exactly as it is,** and no `queue_time` column is - added. Future consumers read the timestamps. See "Alternatives". -- **Clock skew.** Accepted and documented rather than designed around. - Pre-existing across everything derived from compute logging. -- **Per-export rather than per-dataflow stamping.** Correct for today's - single-export dataflows, which the interceptor asserts outright. If multi-output - dataflows land, the fix is to stamp per dataflow and fan out to exports in a - view. -- **No fourth stamp** for the hydration-slot boundary. The interceptor knows it - but runs in environmentd. -- **Crash detection is not compute's job.** A replica cannot report its own death, - and replica lifecycle is already in `mz_cluster_replica_status_history`. -- **Sub-window episodes are unavoidable in the limit.** Mitigated by stamping - event time so that recorded episodes are accurate, and documented as a - 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. - ## 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 "Follow-up work +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 bf048dddc15b2..3c5fee7c35613 100644 --- a/doc/user/content/reference/system-catalog/mz_introspection.md +++ b/doc/user/content/reference/system-catalog/mz_introspection.md @@ -485,6 +485,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 a96caf5281975..3823226b804dd 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -424,11 +424,12 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_CATALOG_SCHEMA, "mz_views", ), - // Required because we added the `mz_cluster_replica_resource_usage` builtin log. - // make_mz_indexes and make_mz_sources inline the builtin-log set as - // VALUES, so adding one changes both MVs' SQL fingerprints. See the NOTE - // above: this version must stay at the workspace's current dev version - // until the change ships. + // Required because we added builtin logs: `mz_cluster_replica_resource_usage` and + // `mz_compute_lifecycle_events_per_worker`. make_mz_indexes and make_mz_sources inline + // the builtin-log set as VALUES, so adding one changes both MVs' SQL fingerprints. A + // replacement step records no fingerprint, so one step per object covers every such + // change at this version. 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, diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs index a3dbda548f476..c8d5495913030 100644 --- a/src/catalog/src/builtin.rs +++ b/src/catalog/src/builtin.rs @@ -1125,6 +1125,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), @@ -2242,6 +2243,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 2e3bdf9764d02..8c440b05cf6f4 100644 --- a/src/catalog/src/builtin/mz_introspection.rs +++ b/src/catalog/src/builtin/mz_introspection.rs @@ -357,6 +357,39 @@ 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: "The lifecycle stages each compute export has reached, with the \ + wallclock instant each was reached at, reported by the worker that \ + observed it.", + 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: Some( + "Both relations are per worker, so the join is on \ + (export_id, worker_id).", + ), + }, + }] + }, + 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 a83d5dd6def40..908251255cd08 100644 --- a/src/catalog/src/durable/transaction.rs +++ b/src/catalog/src/durable/transaction.rs @@ -1022,6 +1022,7 @@ impl<'a> Transaction<'a> { LogVariant::Compute(ComputeLog::OperatorHydrationStatus) => 32, LogVariant::Compute(ComputeLog::PrometheusMetrics) => 33, LogVariant::Compute(ComputeLog::ResourceUsage) => 34, + LogVariant::Compute(ComputeLog::LifecycleEvent) => 35, }; 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 86df318074e85..266697c4e59de 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. @@ -372,6 +374,19 @@ 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("dataflow_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 1569592ede151..834161c6bfb5b 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}; @@ -93,6 +93,13 @@ pub struct ComputeState { /// * Persist sinks store their current frontier in `CollectionState::sink_write_frontier`. /// * Subscribes report their frontiers through the `subscribe_response_buffer`. pub collections: BTreeMap, + /// The exports of each installed dataflow, keyed by dataflow index. + /// + /// Timely mints indices from a per-worker counter that only increases, so an index is never + /// reused within a process. Maintained alongside `collections` by + /// [`ComputeState::insert_collection`] and [`ActiveComputeState::drop_collection`], which is + /// the only reason a dataflow's export set is knowable without scanning every collection. + dataflow_exports: BTreeMap>, /// The traces available for sharing across dataflows. pub traces: TraceManager, /// Shared buffer with SUBSCRIBE operator instances by which they can respond. @@ -208,6 +215,7 @@ impl ComputeState { worker_config: mz_dyncfgs::all_dyncfgs().into(), metrics_registry, workers_per_process, + dataflow_exports: Default::default(), suspended_collections: Default::default(), server_maintenance_interval: Duration::ZERO, init_system_time: mz_ore::now::SYSTEM_TIME(), @@ -216,6 +224,21 @@ impl ComputeState { } } + /// Install the state for a new collection and record it as an export of its dataflow. + /// + /// Returns the state this displaced, which is always a bug in the caller. + fn insert_collection( + &mut self, + id: GlobalId, + collection: CollectionState, + ) -> Option { + self.dataflow_exports + .entry(collection.dataflow_index) + .or_default() + .insert(id); + self.collections.insert(id, collection) + } + /// Return a mutable reference to the identified collection. /// /// Panics if the collection doesn't exist. @@ -624,7 +647,7 @@ impl<'a> ActiveComputeState<'a> { &mut self, dataflow: DataflowDescription, ) { - let dataflow_index = Rc::new(self.timely_worker.next_dataflow_index()); + let dataflow_index = self.timely_worker.next_dataflow_index(); let as_of = dataflow.as_of.clone().unwrap(); let dataflow_expiration = dataflow @@ -689,18 +712,15 @@ impl<'a> ActiveComputeState<'a> { for object_id in dataflow.export_ids() { let is_subscribe_or_copy = subscribe_copy_ids.contains(&object_id); let metrics = self.compute_state.metrics.for_collection(object_id); - let mut collection = CollectionState::new( - Rc::clone(&dataflow_index), - is_subscribe_or_copy, - as_of.clone(), - metrics, - ); + let mut collection = + CollectionState::new(dataflow_index, is_subscribe_or_copy, as_of.clone(), metrics); if let Some(logger) = self.compute_state.compute_logger.clone() { let logging = CollectionLogging::new( object_id, logger, - *dataflow_index, + dataflow_index, + as_of.as_option().copied(), dataflow.import_ids(), ); if starts_immediately { @@ -713,7 +733,7 @@ impl<'a> ActiveComputeState<'a> { lower: as_of.clone(), }); - let existing = self.compute_state.collections.insert(object_id, collection); + let existing = self.compute_state.insert_collection(object_id, collection); if existing.is_some() { error!( id = ?object_id, @@ -745,11 +765,39 @@ impl<'a> ActiveComputeState<'a> { // dataflow can export multiple collections and they all share one suspension token, so the // computation of a dataflow will only start once all its exported collections have been // scheduled. - let suspension_token = self.compute_state.suspended_collections.remove(&id); - drop(suspension_token); + self.compute_state.suspended_collections.remove(&id); - if let Some(collection) = self.compute_state.collections.get(&id) { - if let Some(logging) = &collection.logging { + // Report the start for every export of the dataflow, not just the one this command named. + // Computation begins for all of them at this instant, so crediting each export from its + // own `Schedule` would date the earlier ones to before their dataflow was running and + // overstate the compute time between `started` and `snapshot_complete`. + let Some(collection) = self.compute_state.collections.get(&id) else { + return; + }; + let Some(export_ids) = self + .compute_state + .dataflow_exports + .get(&collection.dataflow_index) + else { + return; + }; + + // An export still holding its token means the dataflow is still suspended, so there is no + // start to report yet. + let still_suspended = export_ids + .iter() + .any(|id| self.compute_state.suspended_collections.contains_key(id)); + if still_suspended { + return; + } + + for export_id in export_ids { + let logging = self + .compute_state + .collections + .get(export_id) + .and_then(|c| c.logging.as_ref()); + if let Some(logging) = logging { logging.set_hydration_start(); } } @@ -824,11 +872,22 @@ impl<'a> ActiveComputeState<'a> { // If this collection is an index, remove its trace. self.compute_state.traces.remove(&id); // If the collection is unscheduled, remove it from the list of waiting collections. + // + // NOTE: dropping the last unscheduled export of a multi-export dataflow releases the + // token here, which unsuspends the dataflow without reporting a hydration start. That + // would charge the surviving exports' queueing time to hydration. Unreachable while every + // dataflow reaching a replica has exactly one export, which + // `SequentialHydration::absorb_command` requires. self.compute_state.suspended_collections.remove(&id); // Drop the dataflow, if all its exports have been dropped. - if let Ok(index) = Rc::try_unwrap(collection.dataflow_index) { - self.timely_worker.drop_dataflow(index); + let index = collection.dataflow_index; + if let Some(exports) = self.compute_state.dataflow_exports.get_mut(&index) { + exports.remove(&id); + if exports.is_empty() { + self.compute_state.dataflow_exports.remove(&index); + self.timely_worker.drop_dataflow(index); + } } // The compute protocol requires us to send a `Frontiers` response with empty frontiers @@ -875,7 +934,6 @@ impl<'a> ActiveComputeState<'a> { storage_log_reader, ); - let dataflow_index = Rc::new(dataflow_index); let mut log_index_ids = config.index_logs; for (log, trace) in traces { // Install trace as maintained index. @@ -888,23 +946,24 @@ impl<'a> ActiveComputeState<'a> { let is_subscribe_or_copy = false; let as_of = Antichain::from_elem(Timestamp::MIN); let metrics = self.compute_state.metrics.for_collection(id); - let mut collection = CollectionState::new( - Rc::clone(&dataflow_index), - is_subscribe_or_copy, - as_of, - metrics, + let mut collection = + CollectionState::new(dataflow_index, is_subscribe_or_copy, as_of.clone(), metrics); + + let logging = CollectionLogging::new( + id, + logger.clone(), + dataflow_index, + as_of.as_option().copied(), + std::iter::empty(), ); - - let logging = - CollectionLogging::new(id, logger.clone(), *dataflow_index, 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 - // without having started. + // start here, or they would sit permanently in the illegal state of having completed + // a snapshot without having started. logging.set_hydration_start(); collection.logging = Some(logging); - let existing = self.compute_state.collections.insert(id, collection); + let existing = self.compute_state.insert_collection(id, collection); if existing.is_some() { error!( id = ?id, @@ -928,7 +987,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 snapshot_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 // collections (database-issues#4701). @@ -957,6 +1017,23 @@ impl<'a> ActiveComputeState<'a> { .allows_reporting(&new_frontier) .then(|| new_frontier.clone()); + // Collect the frontier that measures the dataflow's own progress, for + // `snapshot_complete`. Deliberately not the output frontier collected below, which + // folds in the write frontier and so measures durability instead, and is not uniform + // across workers for a collection that sinks to persist. + // + // A collection without a compute frontier produces its output *by* writing it, an + // index into its own trace, so there the write frontier is the progress. + snapshot_frontier.clear(); + match &collection.compute_probe { + Some(probe) => { + probe.with_frontier(|frontier| { + snapshot_frontier.extend(frontier.iter().copied()) + }); + } + None => snapshot_frontier.clone_from(&new_frontier), + } + // Collect the output frontier and check for progress. // // By default, the output frontier equals the write frontier (which is still stored in @@ -1003,6 +1080,8 @@ impl<'a> ActiveComputeState<'a> { .set_reported_output_frontier(ReportedFrontier::Reported(frontier.clone())); } + collection.observe_snapshot(&snapshot_frontier); + let response = FrontiersResponse { write_frontier: new_write_frontier, input_frontier: new_input_frontier, @@ -1216,6 +1295,15 @@ 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 non-empty batch upper measures progress here. A subscribe reports an + // empty upper both when cancelled and when complete, and completion says nothing + // about the as-of: the sink manufactures the empty upper once + // `up_to <= frontier`, which for `SUBSCRIBE ... UP TO x AS OF x` is reached + // without computing x. A subscribe that did compute through its as-of already + // reported the stage from an earlier, non-empty upper. + if matches!(response, SubscribeResponse::Batch(_)) && !new_frontier.is_empty() { + collection.observe_snapshot(&new_frontier); + } collection.set_reported_output_frontier(ReportedFrontier::Reported(new_frontier)); } else { // Presumably tracking state for this subscribe was already dropped by @@ -1970,10 +2058,9 @@ pub struct CollectionState { reported_frontiers: ReportedFrontiers, /// The index of the dataflow computing this collection. /// - /// Used for dropping the dataflow when the collection is dropped. - /// The Dataflow index is wrapped in an `Rc`s and can be shared between collections, to reflect - /// the possibility that a single dataflow can export multiple collections. - dataflow_index: Rc, + /// A dataflow can compute more than one collection. Which ones is tracked by + /// `ComputeState::dataflow_exports`, which is also what decides when the dataflow is dropped. + dataflow_index: usize, /// Whether this collection is a subscribe or copy-to. /// /// The compute protocol does not allow `Frontiers` responses for subscribe and copy-to @@ -1993,7 +2080,7 @@ pub struct CollectionState { /// Frontier of sink writes. /// /// Only `Some` if the collection is a sink and *not* a subscribe. - pub sink_write_frontier: Option>>>, + sink_write_frontier: Option>>>, /// Frontier probes for every input to the collection. pub input_probes: BTreeMap>, /// A probe reporting the frontier of times through which all collection outputs have been @@ -2005,6 +2092,12 @@ pub struct CollectionState { logging: Option, /// Metrics tracked for this collection. metrics: CollectionMetrics, + /// 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 unfinished again and re-log + /// a stage it already reported. + 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 @@ -2023,7 +2116,7 @@ pub struct CollectionState { impl CollectionState { fn new( - dataflow_index: Rc, + dataflow_index: usize, is_subscribe_or_copy: bool, as_of: Antichain, metrics: CollectionMetrics, @@ -2043,6 +2136,7 @@ impl CollectionState { compute_probe: None, logging: None, metrics, + logged_stages: BTreeSet::new(), read_only_tx, read_only_rx, } @@ -2101,6 +2195,10 @@ 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_snapshot` reports the + /// dataflow-progress reading instead. fn hydrated(&self) -> bool { match &self.reported_frontiers.output_frontier { ReportedFrontier::Reported(frontier) => PartialOrder::less_than(&self.as_of, frontier), @@ -2108,10 +2206,41 @@ 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 this collection's dataflow progress and log the `snapshot_complete` stage the + /// first time that progress has passed the as-of. + /// + /// `progress` must measure the dataflow's own computation rather than the durability of its + /// output. See the comment where it is collected in `report_frontiers`. + /// + /// An empty `progress` counts as complete, matching [`Self::hydrated`]. A caller whose + /// frontier can go empty for a reason other than the dataflow running to its end must exclude + /// that itself, which `process_subscribes` does: a cancelled subscribe reports an empty upper + /// that says nothing about whether the as-of was computed. + fn observe_snapshot(&mut self, progress: &Antichain) { + if PartialOrder::less_than(&self.as_of, progress) { + self.log_stage(LifecycleStage::SnapshotComplete); + } + } + + /// Record the shared frontier a sink publishes its write progress through. + pub(crate) fn set_sink_write_frontier(&mut self, frontier: Rc>>) { + self.sink_write_frontier = Some(frontier); + } + /// Allow writes for this collection. fn allow_writes(&self) { info!( - dataflow_index = *self.dataflow_index, + dataflow_index = self.dataflow_index, export = ?self.logging.as_ref().map(|l| l.export_id()), "allowing writes for dataflow", ); diff --git a/src/compute/src/logging/compute.rs b/src/compute/src/logging/compute.rs index 065d51fceb240..d848bbdeb9048 100644 --- a/src/compute/src/logging/compute.rs +++ b/src/compute/src/logging/compute.rs @@ -10,6 +10,7 @@ //! Logging dataflows for events generated by clusterd. use std::cell::RefCell; +use std::collections::btree_map::Entry; use std::collections::{BTreeMap, BTreeSet}; use std::fmt::{Display, Write}; use std::rc::Rc; @@ -57,6 +58,9 @@ 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. Reported in + /// `details` alongside every lifecycle stage, which are all defined relative to it. + pub as_of: Option, } /// The export for a global id was dropped. @@ -172,6 +176,48 @@ 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. +#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug, Columnar)] +pub enum LifecycleStage { + /// The export's dataflow was installed, still suspended. Logged from [`Export`]. + Installed, + /// The export's dataflow was unsuspended, so hydration work may begin. Logged from + /// [`HydrationStart`]. + Started, + /// The dataflow's own progress frontier passed its as-of, so its snapshot at the as-of is + /// computed. Logged from [`Lifecycle`], as are the stages below. + SnapshotComplete, + /// The export's sink has a batch to mint and read-only mode forbids writing it. + WriteBlockedReadOnly, + /// The export's sink may write, having been reported blocked. + WriteUnblocked, + /// The output shard's upper passed the as-of, so the output is durable through it. + Written, +} + +impl LifecycleStage { + /// The `event` and `reason` this stage is reported as. + fn columns(self) -> (&'static str, Option<&'static str>) { + match self { + Self::Installed => ("installed", None), + Self::Started => ("started", None), + Self::SnapshotComplete => ("snapshot_complete", 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 +284,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 +408,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 +430,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 +449,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 +482,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), @@ -540,6 +593,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 +622,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 +719,39 @@ impl DemuxState { ]) } + /// Pack a lifecycle update key-value for the given export ID and stage. + fn pack_lifecycle_update( + &mut self, + export_id: GlobalId, + dataflow_index: usize, + stage: LifecycleStage, + occurred_at: Duration, + as_of: Option, + ) -> (&RowRef, &RowRef) { + let (event, reason) = stage.columns(); + + // The as-of is 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. + let mut details = Row::default(); + match as_of { + Some(ts) => { + let ts = make_string_datum(ts, &mut self.scratch_string_b); + details.packer().push_dict([("as_of", ts)]); + } + None => details.packer().push_dict([("as_of", Datum::JsonNull)]), + } + + self.lifecycle_packer.pack_slice(&[ + make_string_datum(export_id, &mut self.scratch_string_a), + Datum::UInt64(u64::cast_from(self.worker_id)), + Datum::UInt64(u64::cast_from(dataflow_index)), + Datum::String(event), + epoch_offset_datum(occurred_at), + reason.map_or(Datum::Null, Datum::String), + details.unpack_first(), + ]) + } + /// Pack an import frontier update key-value for the given export ID and dataflow index. fn pack_import_frontier_update( &mut self, @@ -799,10 +888,25 @@ struct ExportState { hydration_timestamps: HydrationTimestamps, /// Hydration status of operators feeding this export. operator_hydration: BTreeMap, + /// The as-of the dataflow maintaining this export was installed with. + /// + /// A dataflow retained across reconciliation logs no new `Export` event, so this stays the + /// install-time value even after the controller re-derives one. That is what the export's + /// logged stages were measured against. + initial_as_of: Option, + /// The lifecycle rows logged for this export so far, keyed by the stage they report. + /// + /// The lifecycle relation is append-only for the life of an export, so the rows are kept to + /// retract exactly what was inserted. + lifecycle_rows: BTreeMap, } impl ExportState { - fn new(dataflow_index: usize, installed_at: Duration) -> Self { + fn new( + dataflow_index: usize, + installed_at: Duration, + initial_as_of: Option, + ) -> Self { Self { dataflow_index, error_count: Diff::ZERO, @@ -814,6 +918,8 @@ impl ExportState { hydrated_at: None, }, operator_hydration: BTreeMap::new(), + initial_as_of, + lifecycle_rows: BTreeMap::new(), } } } @@ -837,6 +943,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 +995,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 +1007,7 @@ impl DemuxHandler<'_, '_, '_> { ExportReference { export_id, dataflow_index, + as_of, }: Ref<'_, Export>, ) { let export_id = Columnar::into_owned(export_id); @@ -910,10 +1019,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 +1038,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 +1076,13 @@ impl DemuxHandler<'_, '_, '_> { .hydration_time .give((datum, ts, Diff::MINUS_ONE)); + // Remove lifecycle logging for this export. + for (key, value) in export.lifecycle_rows.values() { + 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 +1197,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 +1232,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 +1248,84 @@ 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 `snapshot_complete` 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. + /// + /// A stage already logged for this export is ignored, so a caller that cannot cheaply tell + /// whether it has already reported one does not have to. + fn log_lifecycle_at( + &mut self, + export_id: GlobalId, + stage: LifecycleStage, + occurred_at: Duration, + ) { + let ts = self.ts(); + + let Some((initial_as_of, dataflow_index)) = self + .state + .exports + .get(&export_id) + .map(|e| (e.initial_as_of, e.dataflow_index)) + else { + error!(%export_id, ?stage, "lifecycle event for unknown export"); + return; + }; + + let update = { + let (key, value) = self.state.pack_lifecycle_update( + export_id, + dataflow_index, + stage, + occurred_at, + initial_as_of, + ); + (key.to_owned(), value.to_owned()) + }; + + let export = self + .state + .exports + .get_mut(&export_id) + .expect("checked above"); + let Entry::Vacant(entry) = export.lifecycle_rows.entry(stage) else { + return; + }; + let update = entry.insert(update); + + self.output + .lifecycle + .give(((&*update.0, &*update.1), ts, Diff::ONE)); + } + + 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 +1654,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 +1746,17 @@ impl CollectionLogging { })); } + /// Record that the collection reached a stage of its lifecycle. + /// + /// A stage already reported for this export is ignored by the demux, so a caller that cannot + /// cheaply tell whether it has reported one does not have to track it. + 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.rs b/src/compute/src/sink.rs index 002e19738b05e..7851113cb1dd2 100644 --- a/src/compute/src/sink.rs +++ b/src/compute/src/sink.rs @@ -7,6 +7,14 @@ // the Business Source License, use of this software will be governed // by the Apache License, Version 2.0. +use differential_dataflow::hashable::Hashable; +use mz_ore::cast::CastFrom; +use mz_repr::{GlobalId, Timestamp}; +use timely::PartialOrder; +use timely::progress::Antichain; + +use crate::logging::compute::{ComputeEvent, Lifecycle, LifecycleStage, Logger as ComputeLogger}; + mod copy_to_s3_oneshot; #[cfg(feature = "bench")] pub mod correction; @@ -21,3 +29,88 @@ mod materialized_view_v2; mod metric_sink; mod refresh; mod subscribe; + +/// The worker that maintains a persist 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. The election is +/// private to the sink: the frontier leaves it as an input to the controller-visible meet, which +/// needs no owner, and the stages that do need one are reported from `mint` itself. +fn frontier_owner(sink_id: GlobalId, peers: usize) -> usize { + usize::cast_from(sink_id.hashed()) % peers +} + +/// Reports the write lifecycle stages of a persist sink. +/// +/// Only the worker that mints batch descriptions tracks the output shard's upper, so it is the +/// only worker that can report these stages. That is what makes them one report per sink rather +/// than one per worker, and it is why this is held by `mint` rather than by the collection. +/// +/// Each stage is reported at most once. The demux deduplicates as well, so this only keeps +/// repeated observations off the logging channel. +struct WriteStageLogger { + export_id: GlobalId, + /// The as-of `written` is measured against. + as_of: Antichain, + /// `None` when compute logging is disabled. + logger: Option, + blocked: bool, + written: bool, +} + +impl WriteStageLogger { + fn new( + export_id: GlobalId, + as_of: Antichain, + logger: Option, + ) -> Self { + Self { + export_id, + as_of, + logger, + blocked: false, + written: false, + } + } + + /// Report that there is a batch to mint and read-only mode forbids writing it. + /// + /// The caller decides what makes a block worth reporting. Reporting every read-only + /// observation would put a block on essentially every materialized view, since collections + /// start read-only and the controller releases them. + fn blocked(&mut self) { + if !self.blocked { + self.blocked = true; + self.log(LifecycleStage::WriteBlockedReadOnly); + } + } + + /// Report that writing is now permitted, if we reported it blocked. A sink that was never + /// seen to wait has nothing to report here. + fn unblocked(&mut self) { + if self.blocked { + self.log(LifecycleStage::WriteUnblocked); + } + } + + /// Report `written` once the shard's upper passes the as-of. + /// + /// NOTE: this says the output is durable through the as-of, not that this replica wrote it. + /// Every replica reads the same upper back from persist, so it advances on all of them when + /// any one wins the append. + fn observe_persist_frontier(&mut self, frontier: &Antichain) { + if !self.written && PartialOrder::less_than(&self.as_of, frontier) { + self.written = true; + self.log(LifecycleStage::Written); + } + } + + fn log(&self, stage: LifecycleStage) { + if let Some(logger) = &self.logger { + logger.log(&ComputeEvent::Lifecycle(Lifecycle { + export_id: self.export_id, + stage, + })); + } + } +} diff --git a/src/compute/src/sink/materialized_view.rs b/src/compute/src/sink/materialized_view.rs index e5d1a4064c1ba..00b8d7f1de077 100644 --- a/src/compute/src/sink/materialized_view.rs +++ b/src/compute/src/sink/materialized_view.rs @@ -295,7 +295,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.set_sink_write_frontier(sink_frontier); Rc::new((persist_token, mint_token, write_token, append_token)) } @@ -505,7 +505,13 @@ 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::frontier_owner(sink_id, scope.peers()); + + let write_stages = crate::sink::WriteStageLogger::new( + sink_id, + as_of.clone(), + scope.worker().logger_for("materialize/compute"), + ); let sink_frontier = Rc::new(RefCell::new(Antichain::from_elem(Timestamp::MIN))); let shared_frontier = Rc::clone(&sink_frontier); @@ -557,7 +563,7 @@ mod mint { let mut cap_set = CapabilitySet::from_elem(desc_cap); let read_only = *read_only_rx.borrow_and_update(); - let mut state = State::new(sink_id, worker_count, as_of, read_only); + let mut state = State::new(sink_id, worker_count, write_stages, as_of, read_only); // Create a stream that reports advancements of the target shard's frontier and updates // the shared sink frontier. @@ -670,12 +676,15 @@ mod mint { /// /// In read-only mode, minting of batch descriptions is disabled. read_only: bool, + /// Reports the write lifecycle stages of this sink. + write_stages: crate::sink::WriteStageLogger, } impl State { fn new( sink_id: GlobalId, worker_count: usize, + write_stages: crate::sink::WriteStageLogger, as_of: Antichain, read_only: bool, ) -> Self { @@ -692,6 +701,7 @@ mod mint { next_append_worker: 0, last_lower: None, read_only, + write_stages, } } @@ -722,11 +732,18 @@ mod mint { if advance(&mut self.persist_frontier, frontier.borrow()) { self.trace("advanced `persist` frontier"); } + let Self { + write_stages, + persist_frontier, + .. + } = self; + write_stages.observe_persist_frontier(persist_frontier); } fn allow_writes(&mut self) { if self.read_only { self.read_only = false; + self.write_stages.unblocked(); self.trace("disabled read-only mode"); } } @@ -744,6 +761,11 @@ mod mint { PartialOrder::less_than(lower, persist_frontier) }); + // A block only matters once there is a batch to mint. + if self.read_only && desired_ahead && persist_advanced { + self.write_stages.blocked(); + } + if self.read_only || !desired_ahead || !persist_advanced { return None; } diff --git a/src/compute/src/sink/materialized_view_v2.rs b/src/compute/src/sink/materialized_view_v2.rs index 5408d63142025..e3d0545128488 100644 --- a/src/compute/src/sink/materialized_view_v2.rs +++ b/src/compute/src/sink/materialized_view_v2.rs @@ -74,6 +74,7 @@ use tracing::trace; use crate::compute_state::ComputeState; use crate::render::StartSignal; use crate::render::errors::DataflowErrorSer; +use crate::sink::WriteStageLogger; use crate::sink::correction::{ChannelLogging, Correction, CorrectionLogger}; use crate::sink::materialized_view::{ BatchDescription, BatchesStream, DescsStream, DesiredStreams, OkErr, PersistApi, @@ -135,7 +136,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.set_sink_write_frontier(sink_frontier); Rc::new(persist_token) } @@ -165,7 +166,13 @@ 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::frontier_owner(sink_id, scope.peers()); + + let write_stages = WriteStageLogger::new( + sink_id, + as_of.clone(), + scope.worker().logger_for("materialize/compute"), + ); let sink_frontier = Rc::new(RefCell::new(Antichain::from_elem(Timestamp::MIN))); let shared_frontier = Rc::clone(&sink_frontier); @@ -261,6 +268,7 @@ mod mint { state = Some(State::new( sink_id, worker_count, + write_stages, as_of, read_only, persist_rx, @@ -358,12 +366,15 @@ mod mint { /// /// In read-only mode, minting of batch descriptions is disabled. read_only: bool, + /// Reports the write lifecycle stages of this sink. + write_stages: WriteStageLogger, } impl State { fn new( sink_id: GlobalId, worker_count: usize, + write_stages: WriteStageLogger, as_of: Antichain, read_only: bool, persist_rx: mpsc::UnboundedReceiver>, @@ -382,6 +393,7 @@ mod mint { next_append_worker: 0, last_lower: None, read_only, + write_stages, } } @@ -412,6 +424,12 @@ mod mint { if advance(&mut self.persist_frontier, frontier) { self.trace("advanced `persist` frontier"); } + let Self { + write_stages, + persist_frontier, + .. + } = self; + write_stages.observe_persist_frontier(persist_frontier); } /// Drain persist frontier updates from the Tokio task. @@ -457,6 +475,7 @@ mod mint { fn allow_writes(&mut self) { if self.read_only { self.read_only = false; + self.write_stages.unblocked(); self.trace("switched to write mode"); } } @@ -474,6 +493,11 @@ mod mint { PartialOrder::less_than(lower, persist_frontier) }); + // A block only matters once there is a batch to mint. + if self.read_only && desired_ahead && persist_advanced { + self.write_stages.blocked(); + } + if self.read_only || !desired_ahead || !persist_advanced { return None; } diff --git a/src/compute/src/sink/metric_sink.rs b/src/compute/src/sink/metric_sink.rs index 355e74cf7c3a3..2eb1ae4138cc9 100644 --- a/src/compute/src/sink/metric_sink.rs +++ b/src/compute/src/sink/metric_sink.rs @@ -184,9 +184,10 @@ impl<'scope> SinkRender<'scope> for MetricSinkConnection { // Report frontier updates to the `ComputeState`. A metric sink writes to the metrics // registry rather than to a collection, so its "write" frontier is the input frontier it - // has folded through. + // has folded through, not a persist upper, and every worker writes it. Hence not owned: + // no single copy carries progress, and a metric sink reports no write lifecycle stages. let collection = compute_state.expect_collection_mut(sink_id); - collection.sink_write_frontier = Some(sink_frontier); + collection.set_sink_write_frontier(sink_frontier); Some(Rc::new(drop_handle)) } diff --git a/src/pgrepr-consts/src/oid.rs b/src/pgrepr-consts/src/oid.rs index aa820195e6f1a..fa5759bcb2b81 100644 --- a/src/pgrepr-consts/src/oid.rs +++ b/src/pgrepr-consts/src/oid.rs @@ -830,3 +830,4 @@ pub const MV_MZ_METRIC_SINKS_OID: u32 = 17120; pub const INDEX_MZ_METRIC_SINKS_IND_OID: u32 = 17121; pub const TABLE_MZ_OBJECT_HYDRATION_HISTORY_OID: u32 = 17122; pub const LOG_MZ_CLUSTER_REPLICA_RESOURCE_USAGE_OID: u32 = 17123; +pub const LOG_MZ_COMPUTE_LIFECYCLE_EVENTS_PER_WORKER_OID: u32 = 17124; diff --git a/test/sqllogictest/autogenerated/mz_introspection.slt b/test/sqllogictest/autogenerated/mz_introspection.slt index b1f881892146d..676b3a37e4846 100644 --- a/test/sqllogictest/autogenerated/mz_introspection.slt +++ b/test/sqllogictest/autogenerated/mz_introspection.slt @@ -291,6 +291,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 b06437fb06918..a98004baff12d 100644 --- a/test/sqllogictest/catalog_server_explain.slt +++ b/test/sqllogictest/catalog_server_explain.slt @@ -4986,7 +4986,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 (33 rows) + →Constant (34 rows) Source mz_internal.mz_catalog_raw project=(#0..=#2) @@ -5392,7 +5392,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 (57 rows) + →Constant (58 rows) →Arrange (#0{schema_name}) (#0{schema_name}, #1{name}) →Fused with Child Map/Filter/Project Project: #4, #3, #5 @@ -8236,7 +8236,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_builtin_sources"; ---- Explained Query (fast path): - →Constant (57 rows) + →Constant (58 rows) Target cluster: mz_catalog_server @@ -10248,7 +10248,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_ontology_link_types"; ---- Explained Query (fast path): - →Constant (180 rows) + →Constant (181 rows) Target cluster: mz_catalog_server @@ -10315,7 +10315,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 (281 rows) + →Constant (282 rows) →Return →Union →Map/Filter/Project diff --git a/test/sqllogictest/cluster.slt b/test/sqllogictest/cluster.slt index b793028eeabfd..90377ef30f47d 100644 --- a/test/sqllogictest/cluster.slt +++ b/test/sqllogictest/cluster.slt @@ -218,6 +218,13 @@ 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 dataflow_id NULL false +bar mz_compute_lifecycle_events_per_worker mz_compute_lifecycle_events_per_worker_u7_primary_idx 4 event NULL false +bar mz_compute_lifecycle_events_per_worker mz_compute_lifecycle_events_per_worker_u7_primary_idx 5 occurred_at NULL false +bar mz_compute_lifecycle_events_per_worker mz_compute_lifecycle_events_per_worker_u7_primary_idx 6 reason NULL true +bar mz_compute_lifecycle_events_per_worker mz_compute_lifecycle_events_per_worker_u7_primary_idx 7 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 @@ -423,7 +430,7 @@ CREATE CLUSTER test REPLICAS (foo (SIZE 'scale=1,workers=1')); query I SELECT COUNT(name) FROM mz_indexes; ---- -313 +320 statement ok DROP CLUSTER test CASCADE @@ -431,7 +438,7 @@ DROP CLUSTER test CASCADE query T SELECT COUNT(name) FROM mz_indexes; ---- -280 +286 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 e8e5aa17e4007..f9aeb7cd6e0d4 100644 --- a/test/sqllogictest/cockroach/srfs.slt +++ b/test/sqllogictest/cockroach/srfs.slt @@ -1276,6 +1276,48 @@ 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_lifecycle_events_per_worker 7 +mz_compute_lifecycle_events_per_worker 7 +mz_compute_lifecycle_events_per_worker 7 +mz_compute_lifecycle_events_per_worker 7 +mz_compute_lifecycle_events_per_worker 7 +mz_compute_lifecycle_events_per_worker 7 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 e3d7d108b4dcd..fdf674321fc13 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 4d0d5d7ac2a25..3618b98bad721 100644 --- a/test/sqllogictest/information_schema_tables.slt +++ b/test/sqllogictest/information_schema_tables.slt @@ -993,6 +993,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 5a36a7cee42a8..d26bb4035baaf 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)>>>> @@ -157,6 +158,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)> diff --git a/test/sqllogictest/mz_catalog_server_index_accounting.slt b/test/sqllogictest/mz_catalog_server_index_accounting.slt index e8c638c152d60..91e5dcf768e74 100644 --- a/test/sqllogictest/mz_catalog_server_index_accounting.slt +++ b/test/sqllogictest/mz_catalog_server_index_accounting.slt @@ -37,110 +37,111 @@ 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␠[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␠[s770␠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␠[s522␠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␠[s771␠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␠[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␠[s764␠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␠[s619␠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␠[s620␠AS␠"mz_internal"."mz_cluster_replica_name_history"]␠("id") +mz_cluster_reconfigurations_ind CREATE␠INDEX␠"mz_cluster_reconfigurations_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s521␠AS␠"mz_internal"."mz_cluster_reconfigurations"]␠("cluster_id") +mz_cluster_replica_frontiers_ind CREATE␠INDEX␠"mz_cluster_replica_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s765␠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␠[s620␠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␠[s527␠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␠[s528␠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␠[s621␠AS␠"mz_internal"."mz_cluster_replica_name_history"]␠("id") mz_cluster_replica_resource_usage_s2_primary_idx CREATE␠INDEX␠"mz_cluster_replica_resource_usage_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_cluster_replica_resource_usage"␠("process_id",␠"source",␠"metric") -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_cluster_replica_size_internal_ind CREATE␠INDEX␠"mz_cluster_replica_size_internal_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s517␠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␠[s516␠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␠[s529␠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␠[s530␠AS␠"mz_internal"."mz_cluster_replica_statuses"]␠("replica_id") +mz_cluster_replicas_ind CREATE␠INDEX␠"mz_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s526␠AS␠"mz_catalog"."mz_cluster_replicas"]␠("id") +mz_clusters_ind CREATE␠INDEX␠"mz_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s518␠AS␠"mz_catalog"."mz_clusters"]␠("id") +mz_columns_ind CREATE␠INDEX␠"mz_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s489␠AS␠"mz_catalog"."mz_columns"]␠("name") +mz_comments_ind CREATE␠INDEX␠"mz_comments_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s545␠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␠[s745␠AS␠"mz_internal"."mz_compute_dependencies"]␠("dependency_id") +mz_compute_dependencies_ind CREATE␠INDEX␠"mz_compute_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s746␠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␠[s755␠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␠[s756␠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",␠"dataflow_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␠[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␠[s751␠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␠[s750␠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␠[s749␠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_connections_ind CREATE␠INDEX␠"mz_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s524␠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␠[s752␠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␠[s751␠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␠[s750␠AS␠"mz_internal"."mz_console_cluster_utilization_overview"]␠("cluster_id") +mz_databases_ind CREATE␠INDEX␠"mz_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s487␠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␠[s736␠AS␠"mz_internal"."mz_frontiers"]␠("object_id") -mz_hydration_statuses_ind CREATE␠INDEX␠"mz_hydration_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s766␠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␠[s548␠AS␠"mz_catalog"."mz_materialized_views"]␠("id") +mz_frontiers_ind CREATE␠INDEX␠"mz_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s737␠AS␠"mz_internal"."mz_frontiers"]␠("object_id") +mz_hydration_statuses_ind CREATE␠INDEX␠"mz_hydration_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s767␠AS␠"mz_internal"."mz_hydration_statuses"]␠("object_id",␠"replica_id") +mz_indexes_ind CREATE␠INDEX␠"mz_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s490␠AS␠"mz_catalog"."mz_indexes"]␠("id") +mz_kafka_sources_ind CREATE␠INDEX␠"mz_kafka_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s484␠AS␠"mz_catalog"."mz_kafka_sources"]␠("id") +mz_materialized_views_ind CREATE␠INDEX␠"mz_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s549␠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_metric_sinks_ind CREATE␠INDEX␠"mz_metric_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s546␠AS␠"mz_internal"."mz_metric_sinks"]␠("id") -mz_notices_ind CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s843␠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␠[s758␠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␠[s758␠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␠[s756␠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␠[s747␠AS␠"mz_internal"."mz_object_graph_edges"]␠("object_id") -mz_object_history_ind CREATE␠INDEX␠"mz_object_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s559␠AS␠"mz_internal"."mz_object_history"]␠("id") -mz_object_lifetimes_ind CREATE␠INDEX␠"mz_object_lifetimes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s560␠AS␠"mz_internal"."mz_object_lifetimes"]␠("id") -mz_object_transitive_dependencies_ind CREATE␠INDEX␠"mz_object_transitive_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s576␠AS␠"mz_internal"."mz_object_transitive_dependencies"]␠("object_id") -mz_objects_ind CREATE␠INDEX␠"mz_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s556␠AS␠"mz_catalog"."mz_objects"]␠("schema_id") +mz_metric_sinks_ind CREATE␠INDEX␠"mz_metric_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s547␠AS␠"mz_internal"."mz_metric_sinks"]␠("id") +mz_notices_ind CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s844␠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␠[s759␠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␠[s759␠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␠[s757␠AS␠"mz_internal"."mz_object_arrangement_sizes"]␠("replica_id") +mz_object_dependencies_ind CREATE␠INDEX␠"mz_object_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s485␠AS␠"mz_internal"."mz_object_dependencies"]␠("object_id") +mz_object_graph_edges_ind CREATE␠INDEX␠"mz_object_graph_edges_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s748␠AS␠"mz_internal"."mz_object_graph_edges"]␠("object_id") +mz_object_history_ind CREATE␠INDEX␠"mz_object_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s560␠AS␠"mz_internal"."mz_object_history"]␠("id") +mz_object_lifetimes_ind CREATE␠INDEX␠"mz_object_lifetimes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s561␠AS␠"mz_internal"."mz_object_lifetimes"]␠("id") +mz_object_transitive_dependencies_ind CREATE␠INDEX␠"mz_object_transitive_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s577␠AS␠"mz_internal"."mz_object_transitive_dependencies"]␠("object_id") +mz_objects_ind CREATE␠INDEX␠"mz_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s557␠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␠[s720␠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␠[s716␠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␠[s835␠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_recent_activity_log_thinned_ind CREATE␠INDEX␠"mz_recent_activity_log_thinned_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s721␠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␠[s717␠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␠[s836␠AS␠"mz_catalog"."mz_recent_storage_usage"]␠("object_id") +mz_roles_ind CREATE␠INDEX␠"mz_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s508␠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␠[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␠[s604␠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␠[s768␠AS␠"mz_internal"."mz_show_cluster_replicas"]␠("cluster") -mz_show_clusters_ind CREATE␠INDEX␠"mz_show_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s606␠AS␠"mz_internal"."mz_show_clusters"]␠("name") -mz_show_columns_ind CREATE␠INDEX␠"mz_show_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s605␠AS␠"mz_internal"."mz_show_columns"]␠("id") -mz_show_connections_ind CREATE␠INDEX␠"mz_show_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s614␠AS␠"mz_internal"."mz_show_connections"]␠("schema_id") -mz_show_databases_ind CREATE␠INDEX␠"mz_show_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s608␠AS␠"mz_internal"."mz_show_databases"]␠("name") -mz_show_indexes_ind CREATE␠INDEX␠"mz_show_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s618␠AS␠"mz_internal"."mz_show_indexes"]␠("schema_id") -mz_show_materialized_views_ind CREATE␠INDEX␠"mz_show_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s617␠AS␠"mz_internal"."mz_show_materialized_views"]␠("schema_id") -mz_show_roles_ind CREATE␠INDEX␠"mz_show_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s613␠AS␠"mz_internal"."mz_show_roles"]␠("name") -mz_show_schemas_ind CREATE␠INDEX␠"mz_show_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s609␠AS␠"mz_internal"."mz_show_schemas"]␠("database_id") -mz_show_secrets_ind CREATE␠INDEX␠"mz_show_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s607␠AS␠"mz_internal"."mz_show_secrets"]␠("schema_id") -mz_show_sinks_ind CREATE␠INDEX␠"mz_show_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s616␠AS␠"mz_internal"."mz_show_sinks"]␠("schema_id") -mz_show_sources_ind CREATE␠INDEX␠"mz_show_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s615␠AS␠"mz_internal"."mz_show_sources"]␠("schema_id") -mz_show_tables_ind CREATE␠INDEX␠"mz_show_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s610␠AS␠"mz_internal"."mz_show_tables"]␠("schema_id") -mz_show_types_ind CREATE␠INDEX␠"mz_show_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s612␠AS␠"mz_internal"."mz_show_types"]␠("schema_id") -mz_show_views_ind CREATE␠INDEX␠"mz_show_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s611␠AS␠"mz_internal"."mz_show_views"]␠("schema_id") -mz_sink_statistics_ind CREATE␠INDEX␠"mz_sink_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s733␠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␠[s705␠AS␠"mz_internal"."mz_sink_status_history"]␠("sink_id") -mz_sink_statuses_ind CREATE␠INDEX␠"mz_sink_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s706␠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␠[s731␠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␠[s729␠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␠[s707␠AS␠"mz_internal"."mz_source_status_history"]␠("source_id") -mz_source_statuses_ind CREATE␠INDEX␠"mz_source_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s724␠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␠[s740␠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␠[s645␠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␠[s638␠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␠[s655␠AS␠"mz_internal"."pg_authid_core"]␠("rolname") -pg_class_all_databases_ind CREATE␠INDEX␠"pg_class_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s626␠AS␠"mz_internal"."pg_class_all_databases"]␠("relname") -pg_description_all_databases_ind CREATE␠INDEX␠"pg_description_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s635␠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␠[s623␠AS␠"mz_internal"."pg_namespace_all_databases"]␠("nspname") -pg_type_all_databases_ind CREATE␠INDEX␠"pg_type_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s632␠AS␠"mz_internal"."pg_type_all_databases"]␠("oid") +mz_schemas_ind CREATE␠INDEX␠"mz_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s488␠AS␠"mz_catalog"."mz_schemas"]␠("database_id") +mz_secrets_ind CREATE␠INDEX␠"mz_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s523␠AS␠"mz_catalog"."mz_secrets"]␠("name") +mz_show_all_objects_ind CREATE␠INDEX␠"mz_show_all_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s605␠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␠[s769␠AS␠"mz_internal"."mz_show_cluster_replicas"]␠("cluster") +mz_show_clusters_ind CREATE␠INDEX␠"mz_show_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s607␠AS␠"mz_internal"."mz_show_clusters"]␠("name") +mz_show_columns_ind CREATE␠INDEX␠"mz_show_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s606␠AS␠"mz_internal"."mz_show_columns"]␠("id") +mz_show_connections_ind CREATE␠INDEX␠"mz_show_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s615␠AS␠"mz_internal"."mz_show_connections"]␠("schema_id") +mz_show_databases_ind CREATE␠INDEX␠"mz_show_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s609␠AS␠"mz_internal"."mz_show_databases"]␠("name") +mz_show_indexes_ind CREATE␠INDEX␠"mz_show_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s619␠AS␠"mz_internal"."mz_show_indexes"]␠("schema_id") +mz_show_materialized_views_ind CREATE␠INDEX␠"mz_show_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s618␠AS␠"mz_internal"."mz_show_materialized_views"]␠("schema_id") +mz_show_roles_ind CREATE␠INDEX␠"mz_show_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s614␠AS␠"mz_internal"."mz_show_roles"]␠("name") +mz_show_schemas_ind CREATE␠INDEX␠"mz_show_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s610␠AS␠"mz_internal"."mz_show_schemas"]␠("database_id") +mz_show_secrets_ind CREATE␠INDEX␠"mz_show_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s608␠AS␠"mz_internal"."mz_show_secrets"]␠("schema_id") +mz_show_sinks_ind CREATE␠INDEX␠"mz_show_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s617␠AS␠"mz_internal"."mz_show_sinks"]␠("schema_id") +mz_show_sources_ind CREATE␠INDEX␠"mz_show_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s616␠AS␠"mz_internal"."mz_show_sources"]␠("schema_id") +mz_show_tables_ind CREATE␠INDEX␠"mz_show_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s611␠AS␠"mz_internal"."mz_show_tables"]␠("schema_id") +mz_show_types_ind CREATE␠INDEX␠"mz_show_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s613␠AS␠"mz_internal"."mz_show_types"]␠("schema_id") +mz_show_views_ind CREATE␠INDEX␠"mz_show_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s612␠AS␠"mz_internal"."mz_show_views"]␠("schema_id") +mz_sink_statistics_ind CREATE␠INDEX␠"mz_sink_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s734␠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␠[s706␠AS␠"mz_internal"."mz_sink_status_history"]␠("sink_id") +mz_sink_statuses_ind CREATE␠INDEX␠"mz_sink_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s707␠AS␠"mz_internal"."mz_sink_statuses"]␠("id") +mz_sinks_ind CREATE␠INDEX␠"mz_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s500␠AS␠"mz_catalog"."mz_sinks"]␠("id") +mz_source_statistics_ind CREATE␠INDEX␠"mz_source_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s732␠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␠[s730␠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␠[s708␠AS␠"mz_internal"."mz_source_status_history"]␠("source_id") +mz_source_statuses_ind CREATE␠INDEX␠"mz_source_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s725␠AS␠"mz_internal"."mz_source_statuses"]␠("id") +mz_sources_ind CREATE␠INDEX␠"mz_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s493␠AS␠"mz_catalog"."mz_sources"]␠("id") +mz_tables_ind CREATE␠INDEX␠"mz_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s492␠AS␠"mz_catalog"."mz_tables"]␠("schema_id") +mz_types_ind CREATE␠INDEX␠"mz_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s502␠AS␠"mz_catalog"."mz_types"]␠("schema_id") +mz_views_ind CREATE␠INDEX␠"mz_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s501␠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␠[s741␠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␠[s546␠AS␠"mz_internal"."mz_webhook_sources"]␠("id") +pg_attrdef_all_databases_ind CREATE␠INDEX␠"pg_attrdef_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s646␠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␠[s639␠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␠[s656␠AS␠"mz_internal"."pg_authid_core"]␠("rolname") +pg_class_all_databases_ind CREATE␠INDEX␠"pg_class_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s627␠AS␠"mz_internal"."pg_class_all_databases"]␠("relname") +pg_description_all_databases_ind CREATE␠INDEX␠"pg_description_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s636␠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␠[s624␠AS␠"mz_internal"."pg_namespace_all_databases"]␠("nspname") +pg_type_all_databases_ind CREATE␠INDEX␠"pg_type_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s633␠AS␠"mz_internal"."pg_type_all_databases"]␠("oid") # Record all transitive dependencies (tables, sources, views, mvs) of indexes on # the mz_catalog_server cluster. @@ -371,6 +372,13 @@ 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 dataflow_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 ba84f1540219a..5a3206c7bdba4 100644 --- a/test/sqllogictest/oid.slt +++ b/test/sqllogictest/oid.slt @@ -1253,3 +1253,4 @@ SELECT oid, name FROM mz_objects WHERE id LIKE 's%' AND oid < 20000 ORDER BY oid 17121 mz_metric_sinks_ind 17122 mz_object_hydration_history 17123 mz_cluster_replica_resource_usage +17124 mz_compute_lifecycle_events_per_worker diff --git a/test/testdrive/catalog.td b/test/testdrive/catalog.td index d783af6de9cce..a8b1b0a56bd46 100644 --- a/test/testdrive/catalog.td +++ b/test/testdrive/catalog.td @@ -760,6 +760,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 "" @@ -840,7 +841,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%' -280 +286 # 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..db6628b499f61 --- /dev/null +++ b/test/testdrive/compute-lifecycle-events.td @@ -0,0 +1,181 @@ +# 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 `snapshot_complete`. 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 +snapshot_complete 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 = 'snapshot_complete') + >= 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 + +# Every event carries the id of the dataflow maintaining the export, which has to be +# the dataflow `mz_compute_exports_per_worker` reports for that same export and worker. + +> SELECT DISTINCT l.dataflow_id = e.dataflow_id + FROM mz_introspection.mz_compute_lifecycle_events_per_worker l + JOIN mz_introspection.mz_compute_exports_per_worker e + ON (e.export_id = l.export_id AND e.worker_id = l.worker_id) + 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 = 'snapshot_complete' + GROUP BY l.event +snapshot_complete 2 + +# The write stages are ordered against nothing, not the compute stages and not each +# other, so nothing here compares them. `written` reports that the output shard's +# upper passed the as-of, which for a shard that already holds data is true from +# installation, and `apply_refresh` advances the upper of a `REFRESH` materialized +# view before its dataflow has computed anything. What does hold is that `written` carries a +# wallclock instant like every other event. + +> SELECT DISTINCT max(l.occurred_at) BETWEEN now() - '1 hour'::interval AND now() + 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' +true + +# `write_blocked` is reported when the sink has a batch to mint and read-only mode +# forbids writing it, which is a replica 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. This converges rather than holding at every +# instant: `DROP` returns once the catalog row is gone, while the retraction still +# has to reach the replica, be logged, and travel through the introspection +# subscribe, so the retry is doing real work here and a consumer has to tolerate +# the gap. 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', 'snapshot_complete', + '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, where one exists. Only the +# compute stages are ordered. `write_unblocked` still implies `write_blocked`, +# because the sink reports being unblocked only for a sink it reported blocked, but +# that is a causal dependency rather than an ordering of the write stages: `written` +# is unordered against both. + +> 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 ('snapshot_complete' = ANY(events) AND NOT 'started' = ANY(events)) + OR ('write_unblocked' = ANY(events) AND NOT 'write_blocked' = 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 ab3beb5266589..356b10eb1f0e4 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,dataflow_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 ca116ead25529..f7cd906f3fc23 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 354f1df0e5662..365b0e90d8f07 100644 --- a/test/workload-replay/system_catalog_identifiers.txt +++ b/test/workload-replay/system_catalog_identifiers.txt @@ -763,6 +763,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