catalog: add mz_object_hydration_history

Adds the durable table that hydration episodes are recorded into. Nothing writes
it yet, the collector arrives separately.

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 identified by `(object_id, replica_id, installed_at)`, using the
replica-stamped installation time because it is stable across an environmentd
restart. That identity is not declared as a key on the relation: the collector's
anti-join is what keeps it unique, and telling the optimizer a best-effort
sampler's output is unique would turn any duplicate into a silently wrong query
result. None of the comparable history tables declare one either.

No index. An arrangement on the catalog server would hold the whole table, which
grows with objects times replicas times re-hydrations, and nothing queries this
table by key yet.

NOTE: Adding one later is not only an index. `make_mz_indexes` inlines the
builtin index set as VALUES, so a new index changes the `mz_indexes` fingerprint
and needs a `MigrationStep::replacement` for it pinned to the then-current dev
version. A step at a stale version is skipped and the fingerprint check panics at
catalog open.

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.

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,105 @@ 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))
+        .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")