Skip to content

catalog_server: key mz_cluster_replica_frontiers_ind on (object_id, replica_id) #36978

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
antiguru wants to merge 1 commit into from
Closed

catalog_server: key mz_cluster_replica_frontiers_ind on (object_id, replica_id) #36978

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 12 additions & 4 deletions src/catalog/src/builtin/mz_catalog.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2523,14 +2523,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",
Expand Down
29 changes: 16 additions & 13 deletions test/sqllogictest/catalog_server_explain.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ query T multiline
EXPLAIN INDEX "mz_catalog"."mz_cluster_replica_frontiers_ind";
----
mz_catalog.mz_cluster_replica_frontiers_ind:
→Arrange (#0{object_id})
→Arrange (#0{object_id}, #1{replica_id})
→Stream mz_catalog.mz_cluster_replica_frontiers

Source mz_catalog.mz_cluster_replica_frontiers
Expand Down Expand Up @@ -1003,14 +1003,16 @@ mz_internal.mz_hydration_statuses:
→Arranged mz_internal.mz_compute_hydration_times
Key: (#0{replica_id})
→Differential Join %0:l0[#0{id}] » %1:mz_cluster_replica_frontiers[#0{object_id}]
after %1:
Project: #0, #10
Filter: (#11{write_frontier}) IS NULL
Final closure:
Project: #0..=#2
Map: true
→Arranged l0
→Arranged mz_catalog.mz_cluster_replica_frontiers
→Arrange (#0{object_id})
→Fused with Child Map/Filter/Project
Project: #0, #1
Filter: (#2{write_frontier}) IS NULL
→Arranged mz_catalog.mz_cluster_replica_frontiers
Key: (#0{object_id}, #1{replica_id})
cte l2 =
→Differential Join %0:mz_indexes[#0{id}] » %1:l1[#0{object_id}]
→Arranged mz_catalog.mz_indexes
Expand Down Expand Up @@ -1065,10 +1067,9 @@ mz_internal.mz_hydration_statuses:
→Arrange (#0{id}, #2{id})
→Stream l7
cte l9 =
→Differential Join %0:l8[#0{id}, #2{id}] » %1:mz_cluster_replica_frontiers[#0{object_id}, #1{replica_id}]
→Differential Join %1:mz_cluster_replica_frontiers[#0{object_id}, #1{replica_id}] » %0:l8[#0{id}, #2{id}]
→Arranged l8
→Arrange (#0{object_id}, #1{replica_id})
→Arranged mz_catalog.mz_cluster_replica_frontiers
→Arranged mz_catalog.mz_cluster_replica_frontiers
→Return
→Union
→Map/Filter/Project
Expand Down Expand Up @@ -6881,18 +6882,20 @@ Explained Query:
→Arranged mz_internal.mz_compute_hydration_times
Key: (#0{replica_id})
→Differential Join %0:mz_materialized_views[#0{id}] » %1:mz_cluster_replica_frontiers[#0{object_id}]
after %1:
Project: #0, #10
Filter: (#11{write_frontier}) IS NULL
Final closure:
Project: #0..=#3
Map: true, null
→Arranged mz_catalog.mz_materialized_views
→Arranged mz_catalog.mz_cluster_replica_frontiers
→Arrange (#0{object_id})
→Fused with Child Map/Filter/Project
Project: #0, #1
Filter: (#2{write_frontier}) IS NULL
→Arranged mz_catalog.mz_cluster_replica_frontiers
Key: (#0{object_id}, #1{replica_id})

Used Indexes:
- mz_catalog.mz_materialized_views_ind (differential join)
- mz_catalog.mz_cluster_replica_frontiers_ind (differential join)
- mz_catalog.mz_cluster_replica_frontiers_ind (*** full scan ***)
- mz_internal.mz_compute_hydration_times_ind (*** full scan ***)

Target cluster: mz_catalog_server
Expand Down
2 changes: 1 addition & 1 deletion test/sqllogictest/mz_catalog_server_index_accounting.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
2 changes: 1 addition & 1 deletion test/testdrive/indexes.td
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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} ""
Expand Down
Loading