catalog: key mz_cluster_replica_frontiers_ind on (object_id, replica_id)

The `(object_id, replica_id)` join in `mz_hydration_statuses` was building
its own full-relation arrangement of `mz_cluster_replica_frontiers` because
the index was keyed on `object_id` only. That arrangement grows very large
under replica churn (observed ~47 GB on a production mz_catalog_server).
Widening the index key lets the join reuse the index; the remaining
`object_id`-only consumers all filter to `write_frontier IS NULL`, so their
arrangements stay small.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: antiguru

src/catalog/src/builtin/mz_catalog.rs

@@ -2465,14 +2465,22 @@ pub static MZ_CLUSTER_REPLICA_FRONTIERS: LazyLock<BuiltinSource> =
         ontology: None,
     });
 
-pub static MZ_CLUSTER_REPLICA_FRONTIERS_IND: LazyLock<BuiltinIndex> =
-    LazyLock::new(|| BuiltinIndex {
+// Keyed on `(object_id, replica_id)` rather than just `object_id` so that the
+// `(object_id, replica_id)` join in `mz_hydration_statuses` reuses this index
+// instead of building its own full-relation arrangement of
+// `mz_cluster_replica_frontiers` (which grows very large under replica churn;
+// see CLU-112). The remaining `object_id`-only consumers all filter to
+// `write_frontier IS NULL`, so their own arrangements stay small. Do not narrow
+// this key back to `object_id` without re-checking that join.
+pub static MZ_CLUSTER_REPLICA_FRONTIERS_IND: LazyLock<BuiltinIndex> = LazyLock::new(|| {
+    BuiltinIndex {
         name: "mz_cluster_replica_frontiers_ind",
         schema: MZ_CATALOG_SCHEMA,
         oid: oid::INDEX_MZ_CLUSTER_REPLICA_FRONTIERS_IND_OID,
-        sql: "IN CLUSTER mz_catalog_server ON mz_catalog.mz_cluster_replica_frontiers (object_id)",
+        sql: "IN CLUSTER mz_catalog_server ON mz_catalog.mz_cluster_replica_frontiers (object_id, replica_id)",
         is_retained_metrics_object: false,
-    });
+    }
+});
 
 pub static MZ_DEFAULT_PRIVILEGES: LazyLock<BuiltinTable> = LazyLock::new(|| BuiltinTable {
     name: "mz_default_privileges",

test/sqllogictest/mz_catalog_server_index_accounting.slt

@@ -39,7 +39,7 @@ mz_arrangement_records_raw_s2_primary_idx  CREATE␠INDEX␠"mz_arrangement_reco
 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_deployment_lineage_ind  CREATE␠INDEX␠"mz_cluster_deployment_lineage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s747␠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_replica_frontiers_ind  CREATE␠INDEX␠"mz_cluster_replica_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s741␠AS␠"mz_catalog"."mz_cluster_replica_frontiers"]␠("object_id")
+mz_cluster_replica_frontiers_ind  CREATE␠INDEX␠"mz_cluster_replica_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s741␠AS␠"mz_catalog"."mz_cluster_replica_frontiers"]␠("object_id",␠"replica_id")
 mz_cluster_replica_history_ind  CREATE␠INDEX␠"mz_cluster_replica_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s599␠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␠[s509␠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␠[s510␠AS␠"mz_internal"."mz_cluster_replica_metrics"]␠("replica_id")

test/testdrive/indexes.td

@@ -296,7 +296,7 @@ mz_arrangement_batcher_capacity_raw_s2_primary_idx          mz_arrangement_batch
 mz_arrangement_batcher_records_raw_s2_primary_idx           mz_arrangement_batcher_records_raw           mz_catalog_server    {operator_id,worker_id}                     ""
 mz_arrangement_batcher_size_raw_s2_primary_idx              mz_arrangement_batcher_size_raw              mz_catalog_server    {operator_id,worker_id}                     ""
 mz_cluster_deployment_lineage_ind                           mz_cluster_deployment_lineage                mz_catalog_server    {cluster_id}                                ""
-mz_cluster_replica_frontiers_ind                            mz_cluster_replica_frontiers                 mz_catalog_server    {object_id}                                 ""
+mz_cluster_replica_frontiers_ind                            mz_cluster_replica_frontiers                 mz_catalog_server    {object_id,replica_id}                      ""
 mz_cluster_replica_history_ind                              mz_cluster_replica_history                   mz_catalog_server    {dropped_at}                                ""
 mz_cluster_replica_name_history_ind                         mz_cluster_replica_name_history              mz_catalog_server    {id}                                        ""
 mz_cluster_replica_metrics_ind                              mz_cluster_replica_metrics                   mz_catalog_server    {replica_id}                                ""