catalog: add mz_object_hydration_history

Adds the durable table that hydration episodes are recorded into, along with the
index that serves the question users ask of it, how long a given object took to
hydrate. Nothing writes the table yet.

The table is in `mz_internal` because its contents are best effort and its
`status` column will gain values as more hydration events become observable.
An episode is keyed by `(object_id, replica_id, installed_at)`, using the
replica-stamped installation time because it is stable across an environmentd
restart.

Contents are exempt from the bootstrap reset and from forced schema migrations,
since a sampled history cannot be rebuilt from anything else once it is gone.
Durability is best effort in both directions, and the assert added here is a
tripwire so that clearing the table is chosen rather than stumbled into.

Adding a builtin index changes the fingerprint of the `mz_indexes` materialized
view, which inlines the builtin index set as VALUES for exactly that reason, so
this also declares the required replacement migration step.

Ref: SQL-644

Author: aljoscha

doc/user/content/reference/system-catalog/mz_internal.md

@@ -688,6 +688,28 @@ The `mz_object_history` view enriches the [`mz_catalog.mz_objects`](/reference/s
 | `created_at`    | [`timestamp with time zone`]                       | Wall-clock timestamp of when the object was created. `NULL` for built in system objects.                                                                                                |
 | `dropped_at`   | [`timestamp with time zone`]   | Wall-clock timestamp of when the object was dropped. `NULL` for built in system objects or if the object hasn't been dropped.                                              |
 
+## `mz_object_hydration_history`
+
+The `mz_object_hydration_history` table records completed hydration of indexes and
+materialized views, one row per object, replica, and installation. Rows are retained
+for 30 days, and `object_id`, `cluster_id`, and `replica_id` may name objects that no
+longer exist.
+
+Recording is best effort. Only successful hydration is recorded, an episode can be
+missed if the object or its replica goes away before the episode is recorded, and a
+schema change to this table in a future release may clear its contents.
+
+<!-- RELATION_SPEC mz_internal.mz_object_hydration_history -->
+| Field          | Type                         | Meaning                                                                                                                  |
+| -------------- | ---------------------------- | ------------------------------------------------------------------------------------------------------------------------ |
+| `object_id`    | [`text`]                     | The ID of the index or materialized view. May name an object that no longer exists.                                      |
+| `cluster_id`   | [`text`]                     | The ID of the object's cluster.                                                                                          |
+| `replica_id`   | [`text`]                     | The ID of the cluster replica. May name a replica that no longer exists.                                                 |
+| `installed_at` | [`timestamp with time zone`] | When the object's dataflow was installed on the replica.                                                                 |
+| `started_at`   | [`timestamp with time zone`] | When hydration work began, or `NULL` if the replica did not observe a start for this episode. |
+| `finished_at`  | [`timestamp with time zone`] | When hydration finished.                                                                                                 |
+| `status`       | [`text`]                     | The terminal status. Currently always `hydrated`.                                                                        |
+
 ## `mz_object_transitive_dependencies`
 
 The `mz_object_transitive_dependencies` view describes the transitive dependency structure between

src/adapter/src/catalog/open/builtin_schema_migration.rs

@@ -37,7 +37,8 @@ use futures::future::BoxFuture;
 use mz_build_info::{BuildInfo, DUMMY_BUILD_INFO};
 use mz_catalog::builtin::{
     BUILTIN_LOOKUP, Builtin, Fingerprint, MZ_CATALOG_RAW, MZ_CATALOG_RAW_DESCRIPTION,
-    MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY_DESCRIPTION, MZ_STORAGE_USAGE_BY_SHARD,
+    MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY_DESCRIPTION, MZ_OBJECT_HYDRATION_HISTORY,
+    MZ_OBJECT_HYDRATION_HISTORY_DESCRIPTION, MZ_STORAGE_USAGE_BY_SHARD,
     MZ_STORAGE_USAGE_BY_SHARD_DESCRIPTION, RUNTIME_ALTERABLE_FINGERPRINT_SENTINEL,
 };
 use mz_catalog::config::BuiltinItemMigrationConfig;
@@ -390,6 +391,17 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = LazyLock::new(|| {
             MZ_CATALOG_SCHEMA,
             "mz_audit_events",
         ),
+        // Required because we added the `mz_object_hydration_history_ind`
+        // builtin index. make_mz_indexes inlines the builtin-index set as
+        // VALUES, so any add or remove changes its SQL fingerprint and requires
+        // an explicit replacement step. See the NOTE above: this version must
+        // stay at the workspace's current dev version until the change ships.
+        MigrationStep::replacement(
+            "26.39.0-dev.0",
+            CatalogItemType::MaterializedView,
+            MZ_CATALOG_SCHEMA,
+            "mz_indexes",
+        ),
     ]
 });
 
@@ -725,6 +737,17 @@ impl Migration {
                 "mz_object_arrangement_size_history cannot be migrated or else the table will be truncated"
             );
 
+            // Unlike the two tables above, there is no correctness hazard here, a
+            // truncation would only lose history. This is a tripwire so that the
+            // loss is chosen rather than stumbled into: if a schema change to
+            // this table is worth clearing it for, remove this assert along with
+            // the exemption in `plan_forced_migration`, and say in the release
+            // notes that the history restarts.
+            assert_ne!(
+                &*MZ_OBJECT_HYDRATION_HISTORY_DESCRIPTION, object,
+                "migrating mz_object_hydration_history clears it, see the comment above"
+            );
+
             // `mz_catalog_raw` cannot be migrated because it contains the durable catalog and it
             // wouldn't be very durable if we allowed it to be truncated.
             assert_ne!(
@@ -790,9 +813,17 @@ impl Migration {
             .filter(|(_, info)| {
                 use Builtin::*;
                 match info.builtin {
-                    // Filter out the 'mz_storage_usage_by_shard' table since we need to retain
-                    // that info for billing purposes.
-                    Table(table) => **table != *MZ_STORAGE_USAGE_BY_SHARD,
+                    // A forced migration allocates a fresh shard, which discards
+                    // the table's contents. Exclude the tables whose contents
+                    // are the point: storage usage is retained for billing, and
+                    // hydration history cannot be rebuilt from any other source.
+                    // The hydration exemption is best effort, not a guarantee.
+                    // See the tripwire in `plan_migration` for how to give it up
+                    // deliberately.
+                    Table(table) => {
+                        **table != *MZ_STORAGE_USAGE_BY_SHARD
+                            && **table != *MZ_OBJECT_HYDRATION_HISTORY
+                    }
                     MaterializedView(..) => true,
                     Source(source) => **source != *MZ_CATALOG_RAW,
                     Log(..) | View(..) | Type(..) | Func(..) | Index(..) | Connection(..) => false,

src/adapter/src/coord.rs

@@ -98,7 +98,8 @@ use mz_adapter_types::dyncfgs::{
 use mz_auth::password::Password;
 use mz_build_info::BuildInfo;
 use mz_catalog::builtin::{
-    BUILTINS, BUILTINS_STATIC, MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY, MZ_STORAGE_USAGE_BY_SHARD,
+    BUILTINS, BUILTINS_STATIC, MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY, MZ_OBJECT_HYDRATION_HISTORY,
+    MZ_STORAGE_USAGE_BY_SHARD,
 };
 use mz_catalog::config::{AwsPrincipalContext, BuiltinItemMigrationConfig, ClusterReplicaSizeMap};
 use mz_catalog::durable::OpenableDurableCatalogState;
@@ -149,7 +150,7 @@ use mz_secrets::cache::CachingSecretsReader;
 use mz_secrets::{SecretsController, SecretsReader};
 use mz_sql::ast::{Raw, Statement};
 use mz_sql::catalog::{CatalogCluster, EnvironmentId};
-use mz_sql::names::{QualifiedItemName, ResolvedIds, SchemaSpecifier};
+use mz_sql::names::{QualifiedItemName, ResolvedIds};
 use mz_sql::optimizer_metrics::OptimizerMetrics;
 use mz_sql::plan::{
     self, AlterSinkPlan, ConnectionDetails, CreateConnectionPlan, HirRelationExpr,
@@ -3064,29 +3065,19 @@ impl Coordinator {
         debug!("coordinator init: resetting system tables");
         let read_ts = self.get_local_read_ts().await;
 
-        // Filter out tables whose contents must survive restarts:
-        // 'mz_storage_usage_by_shard' for billing, and
-        // 'mz_object_arrangement_size_history', which accumulates history that
-        // is pruned by its own retention period instead.
-        let mz_storage_usage_by_shard_schema: SchemaSpecifier = self
-            .catalog()
-            .resolve_system_schema(MZ_STORAGE_USAGE_BY_SHARD.schema)
-            .into();
-        let arrangement_size_history_schema: SchemaSpecifier = self
-            .catalog()
-            .resolve_system_schema(MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY.schema)
-            .into();
-        let is_retained_across_restarts = |meta: &TableMetadata| -> bool {
-            (meta.name.item == MZ_STORAGE_USAGE_BY_SHARD.name
-                && meta.name.qualifiers.schema_spec == mz_storage_usage_by_shard_schema)
-                || (meta.name.item == MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY.name
-                    && meta.name.qualifiers.schema_spec == arrangement_size_history_schema)
-        };
+        let retained_across_restarts = BTreeSet::from([
+            self.catalog()
+                .resolve_builtin_table(&MZ_STORAGE_USAGE_BY_SHARD),
+            self.catalog()
+                .resolve_builtin_table(&MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY),
+            self.catalog()
+                .resolve_builtin_table(&MZ_OBJECT_HYDRATION_HISTORY),
+        ]);
 
         let mut retraction_tasks = Vec::new();
         let system_tables: Vec<_> = table_metas
             .iter()
-            .filter(|meta| meta.id.is_system() && !is_retained_across_restarts(meta))
+            .filter(|meta| meta.id.is_system() && !retained_across_restarts.contains(&meta.id))
             .collect();
 
         for system_table in system_tables {

src/catalog/src/builtin.rs

@@ -836,6 +836,14 @@ pub static MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY_DESCRIPTION: LazyLock<SystemObject
         object_type: CatalogItemType::Table,
         object_name: MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY.name.to_string(),
     });
+
+/// Identifies [`MZ_OBJECT_HYDRATION_HISTORY`] for the schema-migration guard.
+pub static MZ_OBJECT_HYDRATION_HISTORY_DESCRIPTION: LazyLock<SystemObjectDescription> =
+    LazyLock::new(|| SystemObjectDescription {
+        schema_name: MZ_OBJECT_HYDRATION_HISTORY.schema.to_string(),
+        object_type: CatalogItemType::Table,
+        object_name: MZ_OBJECT_HYDRATION_HISTORY.name.to_string(),
+    });
 pub const MZ_SYSTEM_ROLE: BuiltinRole = BuiltinRole {
     id: MZ_SYSTEM_ROLE_ID,
     name: SYSTEM_USER_NAME,
@@ -1463,6 +1471,13 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::ne
         Builtin::View(&MZ_INDEX_ADVICE),
         Builtin::View(&MZ_MCP_DATA_PRODUCTS),
         Builtin::View(&MZ_MCP_DATA_PRODUCT_DETAILS),
+        // NOTE: This list is a dependency order, and in a fresh environment it
+        // is also the order builtin ids are handed out in. Appending keeps a new
+        // builtin from shifting the ids of the ones before it, which is what the
+        // catalog goldens record. An existing environment is unaffected either
+        // way: its builtins keep the ids stored under their names.
+        Builtin::Table(&MZ_OBJECT_HYDRATION_HISTORY),
+        Builtin::Index(&MZ_OBJECT_HYDRATION_HISTORY_IND),
     ];
 
     builtin_items.extend(notice::builtins());

src/catalog/src/builtin/mz_internal.rs

@@ -4965,6 +4965,83 @@ pub static MZ_OBJECT_ARRANGEMENT_SIZE_HISTORY_TS_IND: LazyLock<BuiltinIndex> =
         is_retained_metrics_object: true,
     });
 
+/// Completed hydration episodes, one row per object, replica, and installation.
+///
+/// Durability here is best effort, in both directions. We keep the contents out
+/// of the bootstrap reset and out of forced schema migrations, because a sampled
+/// history cannot be rebuilt from anything else once it is gone. But we do not
+/// promise to preserve it forever: a future schema change may be worth more than
+/// the accumulated rows, and clearing the table is an acceptable price to pay for
+/// one. Nothing may depend on a row still being here.
+pub static MZ_OBJECT_HYDRATION_HISTORY: LazyLock<BuiltinTable> = LazyLock::new(|| BuiltinTable {
+    name: "mz_object_hydration_history",
+    schema: MZ_INTERNAL_SCHEMA,
+    oid: oid::TABLE_MZ_OBJECT_HYDRATION_HISTORY_OID,
+    desc: RelationDesc::builder()
+        .with_column("object_id", SqlScalarType::String.nullable(false))
+        .with_column("cluster_id", SqlScalarType::String.nullable(false))
+        .with_column("replica_id", SqlScalarType::String.nullable(false))
+        .with_column(
+            "installed_at",
+            SqlScalarType::TimestampTz { precision: None }.nullable(false),
+        )
+        .with_column(
+            "started_at",
+            SqlScalarType::TimestampTz { precision: None }.nullable(true),
+        )
+        .with_column(
+            "finished_at",
+            SqlScalarType::TimestampTz { precision: None }.nullable(true),
+        )
+        .with_column("status", SqlScalarType::String.nullable(false))
+        .with_key(vec![0, 2, 3])
+        .finish(),
+    column_comments: BTreeMap::from_iter([
+        (
+            "object_id",
+            "The ID of the index or materialized view. May name an object that no longer exists.",
+        ),
+        ("cluster_id", "The ID of the object's cluster."),
+        (
+            "replica_id",
+            "The ID of the cluster replica. May name a replica that no longer exists.",
+        ),
+        (
+            "installed_at",
+            "When the object's dataflow was installed on the replica.",
+        ),
+        (
+            "started_at",
+            "When hydration work began, or `NULL` if the replica did not observe a start for this episode.",
+        ),
+        ("finished_at", "When hydration finished."),
+        (
+            "status",
+            "The terminal status. Currently always `hydrated`.",
+        ),
+    ]),
+    is_retained_metrics_object: true,
+    access: vec![PUBLIC_SELECT],
+    // No ontology links: a history row deliberately outlives the object and the
+    // replica it describes, so a foreign key to either would dangle for exactly
+    // the rows that make this table worth keeping.
+    ontology: None,
+});
+
+pub static MZ_OBJECT_HYDRATION_HISTORY_IND: LazyLock<BuiltinIndex> =
+    LazyLock::new(|| BuiltinIndex {
+        name: "mz_object_hydration_history_ind",
+        schema: MZ_INTERNAL_SCHEMA,
+        oid: oid::INDEX_MZ_OBJECT_HYDRATION_HISTORY_IND_OID,
+        // Keyed for the question users ask of this table, "how long did this
+        // object take to hydrate". The collector's own anti-join looks the same
+        // but cannot use this arrangement: its subscribe runs on the targeted
+        // user replica, where the index does not exist.
+        sql: "IN CLUSTER mz_catalog_server
+ON mz_internal.mz_object_hydration_history (object_id)",
+        is_retained_metrics_object: true,
+    });
+
 pub static MZ_COMPUTE_HYDRATION_STATUSES: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
     name: "mz_compute_hydration_statuses",
     schema: MZ_INTERNAL_SCHEMA,

src/pgrepr-consts/src/oid.rs

@@ -822,3 +822,5 @@ pub const FUNC_PARSE_CONNECTION_DETAILS_OID: u32 = 17112;
 pub const FUNC_MZ_AWS_ACCOUNT_ID_OID: u32 = 17113;
 pub const FUNC_MZ_AWS_EXTERNAL_ID_PREFIX_OID: u32 = 17114;
 pub const FUNC_MZ_AWS_CONNECTION_ROLE_ARN_OID: u32 = 17115;
+pub const TABLE_MZ_OBJECT_HYDRATION_HISTORY_OID: u32 = 17116;
+pub const INDEX_MZ_OBJECT_HYDRATION_HISTORY_IND_OID: u32 = 17117;

test/sqllogictest/autogenerated/mz_internal.slt

@@ -382,6 +382,17 @@ object_type  text  The␠type␠of␠the␠object:␠one␠of␠`table`,␠`sour
 created_at  timestamp␠with␠time␠zone  Wall-clock␠timestamp␠of␠when␠the␠object␠was␠created.␠`NULL`␠for␠built␠in␠system␠objects.
 dropped_at  timestamp␠with␠time␠zone  Wall-clock␠timestamp␠of␠when␠the␠object␠was␠dropped.␠`NULL`␠for␠built␠in␠system␠objects␠or␠if␠the␠object␠hasn't␠been␠dropped.
 
+query TTT
+SELECT name, type, comment FROM objects WHERE schema = 'mz_internal' AND object = 'mz_object_hydration_history' ORDER BY position
+----
+object_id  text  The␠ID␠of␠the␠index␠or␠materialized␠view.␠May␠name␠an␠object␠that␠no␠longer␠exists.
+cluster_id  text  The␠ID␠of␠the␠object's␠cluster.
+replica_id  text  The␠ID␠of␠the␠cluster␠replica.␠May␠name␠a␠replica␠that␠no␠longer␠exists.
+installed_at  timestamp␠with␠time␠zone  When␠the␠object's␠dataflow␠was␠installed␠on␠the␠replica.
+started_at  timestamp␠with␠time␠zone  When␠hydration␠work␠began,␠or␠`NULL`␠if␠the␠replica␠did␠not␠observe␠a␠start␠for␠this␠episode.
+finished_at  timestamp␠with␠time␠zone  When␠hydration␠finished.
+status  text  The␠terminal␠status.␠Currently␠always␠`hydrated`.
+
 query TTT
 SELECT name, type, comment FROM objects WHERE schema = 'mz_internal' AND object = 'mz_object_transitive_dependencies' ORDER BY position
 ----
@@ -830,6 +841,7 @@ mz_object_dependencies
 mz_object_fully_qualified_names
 mz_object_global_ids
 mz_object_history
+mz_object_hydration_history
 mz_object_lifetimes
 mz_object_oid_alias
 mz_object_transitive_dependencies

test/sqllogictest/catalog_server_explain.slt

@@ -1664,6 +1664,19 @@ Target cluster: mz_catalog_server
 
 EOF
 
+query T multiline
+EXPLAIN INDEX "mz_internal"."mz_object_hydration_history_ind";
+----
+mz_internal.mz_object_hydration_history_ind:
+  →Arrange (#0{object_id})
+    →Stream mz_internal.mz_object_hydration_history
+
+Source mz_internal.mz_object_hydration_history
+
+Target cluster: mz_catalog_server
+
+EOF
+
 query T multiline
 EXPLAIN INDEX "mz_internal"."mz_object_lifetimes_ind";
 ----
@@ -4811,7 +4824,7 @@ mz_catalog.mz_indexes:
           Project: #2, #0, #1, #3, #4, #7, #5, #6
           Map: "s1"
       →Arrange (#1{name}) (#2{on_schema}, #3{on_name})
-        →Constant (80 rows)
+        →Constant (81 rows)
       →Arrange (empty key) (#1{name})
         →Fused with Child Map/Filter/Project
           Project: #4, #3

test/sqllogictest/cluster.slt

@@ -420,15 +420,15 @@ CREATE CLUSTER test REPLICAS (foo (SIZE 'scale=1,workers=1'));
 query I
 SELECT COUNT(name) FROM mz_indexes;
 ----
-304
+305
 
 statement ok
 DROP CLUSTER test CASCADE
 
 query T
 SELECT COUNT(name) FROM mz_indexes;
 ----
-272
+273
 
 simple conn=mz_system,user=mz_system
 ALTER CLUSTER quickstart OWNER TO materialize

test/sqllogictest/cockroach/srfs.slt

@@ -1473,6 +1473,7 @@ mz_object_arrangement_size_history  4
 mz_object_arrangement_sizes  1
 mz_object_dependencies  1
 mz_object_history  1
+mz_object_hydration_history  1
 mz_object_lifetimes  1
 mz_object_transitive_dependencies  1
 mz_objects  3