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, with one row for each time an object hydrated on a replica. 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;
@@ -725,6 +726,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 +802,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,7 @@ 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),
+        Builtin::Table(&MZ_OBJECT_HYDRATION_HISTORY),
     ];
 
     builtin_items.extend(notice::builtins());

src/catalog/src/builtin/mz_internal.rs

@@ -4965,6 +4965,106 @@ 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.
+///
+/// Exempt from the bootstrap reset and from forced schema migrations, since the
+/// contents cannot be rebuilt from anything else. Clearing them for a schema
+/// change is still allowed, see the tripwire in `plan_migration`.
+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`.",
+        ),
+    ]),
+    // Not a retained-metrics object: that would pin a 30 day compaction window,
+    // and our history lives in the rows, which the retention sweep retracts on
+    // its own schedule. Nothing reads this table at an old timestamp.
+    is_retained_metrics_object: false,
+    access: vec![PUBLIC_SELECT],
+    ontology: Some(Ontology {
+        entity_name: "object_hydration_event",
+        description: "Completed hydration of an index or materialized view on a replica",
+        // NOTE: These references outlive what they point at. A row deliberately
+        // survives the object and the replica it describes, so resolving one
+        // against the catalog can come up empty.
+        links: &const {
+            [
+                OntologyLink {
+                    name: "hydration_of_object",
+                    target: "object",
+                    properties: LinkProperties::fk_typed(
+                        "object_id",
+                        "id",
+                        Cardinality::ManyToOne,
+                        mz_repr::SemanticType::CatalogItemId,
+                    ),
+                },
+                OntologyLink {
+                    name: "hydrated_on_cluster",
+                    target: "cluster",
+                    properties: LinkProperties::fk("cluster_id", "id", Cardinality::ManyToOne),
+                },
+                OntologyLink {
+                    name: "hydrated_on_replica",
+                    target: "replica",
+                    properties: LinkProperties::fk_typed(
+                        "replica_id",
+                        "id",
+                        Cardinality::ManyToOne,
+                        mz_repr::SemanticType::CatalogItemId,
+                    ),
+                },
+            ]
+        },
+        column_semantic_types: &[
+            ("object_id", SemanticType::CatalogItemId),
+            ("cluster_id", SemanticType::ClusterId),
+            ("replica_id", SemanticType::ReplicaId),
+        ],
+    }),
+});
+
 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,4 @@ 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;

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

@@ -9969,7 +9969,7 @@ query T multiline
 EXPLAIN SELECT * FROM "mz_internal"."mz_ontology_entity_types";
 ----
 Explained Query (fast path):
-  →Constant (134 rows)
+  →Constant (135 rows)
 
 Target cluster: mz_catalog_server
 
@@ -9979,7 +9979,7 @@ query T multiline
 EXPLAIN SELECT * FROM "mz_internal"."mz_ontology_link_types";
 ----
 Explained Query (fast path):
-  →Constant (173 rows)
+  →Constant (176 rows)
 
 Target cluster: mz_catalog_server
 
@@ -9993,7 +9993,7 @@ Explained Query:
     cte l0 =
       →Differential Join %1:mz_schemas[#0{id}] » %2:mz_objects[#2{schema_id}] » %0[#0{schema_name}, #1{table_name}] » %3:mz_columns[#0{id}]
         →Arrange (#0{schema_name}, #1{table_name})
-          →Constant (134 rows)
+          →Constant (135 rows)
         →Arrange (#0{id})
           →Fused with Child Map/Filter/Project
             Project: #1, #3
@@ -10046,7 +10046,7 @@ Explained Query:
       →Differential Join %0:l4[#0{entity_name}, #1{name}] » %1[#0{entity_name}, #1{column_name}]
         →Arranged l4
         →Arrange (#0{entity_name}, #1{column_name})
-          →Constant (272 rows)
+          →Constant (275 rows)
   →Return
     →Union
       →Map/Filter/Project

test/sqllogictest/information_schema_tables.slt

@@ -497,6 +497,10 @@ mz_object_history
 VIEW
 materialize
 mz_internal
+mz_object_hydration_history
+BASE TABLE
+materialize
+mz_internal
 mz_object_lifetimes
 VIEW
 materialize

test/sqllogictest/mz_catalog_server_index_accounting.slt

@@ -83,7 +83,7 @@ mz_message_batch_counts_received_raw_s2_primary_idx  CREATE␠INDEX␠"mz_messag
 mz_message_batch_counts_sent_raw_s2_primary_idx  CREATE␠INDEX␠"mz_message_batch_counts_sent_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_batch_counts_sent_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id")
 mz_message_counts_received_raw_s2_primary_idx  CREATE␠INDEX␠"mz_message_counts_received_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_counts_received_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id")
 mz_message_counts_sent_raw_s2_primary_idx  CREATE␠INDEX␠"mz_message_counts_sent_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_counts_sent_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id")
-mz_notices_ind  CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s835␠AS␠"mz_internal"."mz_notices"]␠("id")
+mz_notices_ind  CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s836␠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␠[s753␠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␠[s753␠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␠[s751␠AS␠"mz_internal"."mz_object_arrangement_sizes"]␠("replica_id")