catalog_server: key mz_cluster_replica_frontiers_ind on (object_id, replica_id)

[antiguru]

Motivation

Fixes CPU-112. On mz_catalog_server, the mz_hydration_statuses index dataflow joins mz_cluster_replica_frontiers on (object_id, replica_id). Because mz_cluster_replica_frontiers_ind was keyed on object_id only, that join built its own full-relation arrangement of mz_cluster_replica_frontiers. Under replica churn this arrangement accumulates a lot of uncompacted history — observed at ~47 GB on a production mz_catalog_server replica via continuous profiling.

Description

Widen the index key to (object_id, replica_id) so the join reuses the index instead of duplicating the relation into its own arrangement. The other consumers that join on object_id only (mz_hydration_statuses index/MV branches and mz_compute_hydration_statuses) all filter to write_frontier IS NULL, so when they build their own object_id arrangements those stay small. Net effect on mz_catalog_server: two full-relation arrangements of mz_cluster_replica_frontiers become one (the index) plus two small filtered ones. A comment on the index records the rationale so the key isn't naively narrowed back.

Caveat: a (object_id, replica_id) index cannot serve an object_id-only equality lookup (Differential arrangements key on the full tuple), so an ad-hoc point lookup such as SELECT ... FROM mz_cluster_replica_frontiers WHERE object_id = X (without replica_id) no longer uses this index and falls back to a scan. This is minor — the relation is small per object — and is the intended trade-off, but it's a visible behavior change beyond the maintained dataflows.

Verification

The plan impact is captured by the catalog_server_explain.slt snapshot (#36995), regenerated here. It shows exactly three changed plans, all expected:

• mz_cluster_replica_frontiers_ind — the new key.
• mz_hydration_statuses_ind — l9 reuses the index (its fresh (object_id, replica_id) arrangement is gone); l1 builds a small write_frontier IS NULL-filtered object_id arrangement.
• mz_compute_hydration_statuses — same small-filtered-arrangement shift; this is an unindexed SELECT * view, so it is only surfaced by the snapshot, not by an EXPLAIN INDEX sweep.

No other mz_catalog_server object's plan changes. mz_catalog_server_index_accounting.slt and indexes.td are updated for the new key; sqllogictest and cargo clippy -p mz-catalog pass.

🤖 Generated with Claude Code

[antiguru]

Query plans: before (object_id index) vs after ((object_id, replica_id) index)

Collected via EXPLAIN INDEX on a local build. Only the index itself and mz_hydration_statuses_ind change; all other 106 indexes and 13 materialized views on mz_catalog_server are byte-identical.

mz_catalog.mz_cluster_replica_frontiers_ind

Before

mz_catalog.mz_cluster_replica_frontiers_ind:
  →Arrange (#0{object_id})
    →Stream mz_catalog.mz_cluster_replica_frontiers
Source mz_catalog.mz_cluster_replica_frontiers
Target cluster: mz_catalog_server

After

mz_catalog.mz_cluster_replica_frontiers_ind:
  →Arrange (#0{object_id}, #1{replica_id})
    →Stream mz_catalog.mz_cluster_replica_frontiers
Source mz_catalog.mz_cluster_replica_frontiers
Target cluster: mz_catalog_server

mz_internal.mz_hydration_statuses_ind

The relevant change is in cte l1 and cte l9 (the mz_cluster_replica_frontiers arrangements). Before: l9 builds a fresh Arrange (object_id, replica_id) over the full relation. After: l9 reads Arranged mz_cluster_replica_frontiers directly (reuses the index), and l1 builds a small object_id arrangement filtered to write_frontier IS NULL.

Before

mz_internal.mz_hydration_statuses_ind:
  →Arrange (#0{object_id}, #1{replica_id})
    →Stream mz_internal.mz_hydration_statuses
mz_internal.mz_hydration_statuses:
  →With
    cte l0 =
      →Arranged mz_catalog.mz_materialized_views
    cte l1 =
      →Arrange (#0{object_id})
        →Union
          →Fused with Child Map/Filter/Project
            Project: #1, #0, #3
            Map: (#2{time_ns}) IS NOT NULL
              →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
    cte l2 =
      →Differential Join %0:mz_indexes[#0{id}] » %1:l1[#0{object_id}]
        →Arranged mz_catalog.mz_indexes
        →Arranged l1
    cte l3 =
      →Differential Join %0:l0[#0{id}] » %1:l1[#0{object_id}]
        →Arranged l0
        →Arranged l1
    cte l4 =
      →Differential Join %0:mz_sources[#0{id}] » %1:mz_source_statistics[#0{id}]
        after %1:
          Project: #0, #15, #16
          Filter: (#4{type} != "webhook") AND (#10{cluster_id}) IS NOT NULL
        →Arranged mz_catalog.mz_sources
        →Arrange (#0{id})
          →Fused with Child Map/Filter/Project
            Project: #0, #1, #8
              →Arranged mz_internal.mz_source_statistics
                Key: (#0{id}, #1{replica_id})
    cte l5 =
      →Arranged mz_catalog.mz_sinks
    cte l6 =
      →Differential Join %0:l5[#0{id}] » %1:mz_sink_statuses[#0{id}]
        →Arranged l5
        →Arranged mz_internal.mz_sink_statuses
    cte l7 =
      →Differential Join %0[#1{cluster_id}] » %1:mz_cluster_replicas[#1{cluster_id}]
        →Arrange (#1{cluster_id})
          →Union
            →Map/Filter/Project
              Project: #0..=#2
              Map: null
                →Consolidating Union
                  →Negate Diffs
                    →Differential Join %0:l5[#0{id}] » %1[#0]
                      →Arranged l5
                      →Distinct GroupAggregate
                        →Fused with Child Map/Filter/Project
                          Project: #0
                            →Read l6
                  →Fused with Child Map/Filter/Project
                    Project: #0, #11
                      →Arranged mz_catalog.mz_sinks
                        Key: (#0{id})
            →Stream l6
        →Arrange (#1{cluster_id})
          →Fused with Child Map/Filter/Project
            Project: #0, #2
              →Arranged mz_catalog.mz_cluster_replicas
                Key: (#0{id})
    cte l8 =
      →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}]
        →Arranged l8
        →Arrange (#0{object_id}, #1{replica_id})
          →Arranged mz_catalog.mz_cluster_replica_frontiers
  →Return
    →Union
      →Map/Filter/Project
        Project: #0, #1, #3
        Map: coalesce(#2{hydrated}, false)
          →Union
            →Map/Filter/Project
              Project: #0..=#2
              Map: null, null
                →Consolidating Union
                  →Negate Diffs
                    →Unarranged Raw Stream
                      →Distinct GroupAggregate
                        →Fused with Child Map/Filter/Project
                          Project: #0
                            →Read l2
                  →Fused with Child Map/Filter/Project
                    Project: #0
                      →Arranged mz_catalog.mz_indexes
                        Key: (#0{id})
            →Stream l2
      →Map/Filter/Project
        Project: #0, #1, #3
        Map: coalesce(#2{hydrated}, false)
          →Union
            →Map/Filter/Project
              Project: #0..=#2
              Map: null, null
                →Consolidating Union
                  →Negate Diffs
                    →Unarranged Raw Stream
                      →Distinct GroupAggregate
                        →Fused with Child Map/Filter/Project
                          Project: #0
                            →Read l3
                  →Fused with Child Map/Filter/Project
                    Project: #0
                      →Arranged mz_catalog.mz_materialized_views
                        Key: (#0{id})
            →Stream l3
      →Map/Filter/Project
        Project: #0, #1, #3
        Map: (#2{rehydration_latency}) IS NOT NULL
          →Union
            →Map/Filter/Project
              Project: #0..=#2
              Map: null, null
                →Consolidating Union
                  →Negate Diffs
                    →Unarranged Raw Stream
                      →Distinct GroupAggregate
                        →Fused with Child Map/Filter/Project
                          Project: #0
                            →Read l4
                  →Fused with Child Map/Filter/Project
                    Project: #0
                    Filter: (#4{type} != "webhook") AND (#10{cluster_id}) IS NOT NULL
                      →Arranged mz_catalog.mz_sources
                        Key: (#0{id})
            →Stream l4
      →Map/Filter/Project
        Project: #0, #2, #4
        Map: ((#1{status} = "running") AND (coalesce(#3{write_frontier}, 0) > 0))
          →Union
            →Map/Filter/Project
              Project: #0..=#3
              Map: null
                →Consolidating Union
                  →Negate Diffs
                    →Differential Join %1[#0, #1] » %0:l8[#0{id}, #2{id}]
                      →Arranged l8
                      →Distinct GroupAggregate
                        →Fused with Child Map/Filter/Project
                          Project: #0, #2
                            →Read l9
                  →Stream l7
            →Stream l9
Used Indexes:
  - mz_internal.mz_source_statistics_ind (*** full scan ***)
  - mz_catalog.mz_indexes_ind (*** full scan ***, differential join)
  - mz_catalog.mz_sources_ind (*** full scan ***, differential join)
  - mz_catalog.mz_sinks_ind (*** full scan ***, differential join)
  - mz_catalog.mz_materialized_views_ind (*** full scan ***, differential join)
  - mz_internal.mz_sink_statuses_ind (differential join)
  - mz_catalog.mz_cluster_replicas_ind (*** full scan ***)
  - mz_catalog.mz_cluster_replica_frontiers_ind (*** full scan ***, differential join)
  - mz_internal.mz_compute_hydration_times_ind (*** full scan ***)
Target cluster: mz_catalog_server

After

mz_internal.mz_hydration_statuses_ind:
  →Arrange (#0{object_id}, #1{replica_id})
    →Stream mz_internal.mz_hydration_statuses
mz_internal.mz_hydration_statuses:
  →With
    cte l0 =
      →Arranged mz_catalog.mz_materialized_views
    cte l1 =
      →Arrange (#0{object_id})
        →Union
          →Fused with Child Map/Filter/Project
            Project: #1, #0, #3
            Map: (#2{time_ns}) IS NOT NULL
              →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}]
            Final closure:
              Project: #0..=#2
              Map: true
            →Arranged l0
            →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
        →Arranged l1
    cte l3 =
      →Differential Join %0:l0[#0{id}] » %1:l1[#0{object_id}]
        →Arranged l0
        →Arranged l1
    cte l4 =
      →Differential Join %0:mz_sources[#0{id}] » %1:mz_source_statistics[#0{id}]
        after %1:
          Project: #0, #15, #16
          Filter: (#4{type} != "webhook") AND (#10{cluster_id}) IS NOT NULL
        →Arranged mz_catalog.mz_sources
        →Arrange (#0{id})
          →Fused with Child Map/Filter/Project
            Project: #0, #1, #8
              →Arranged mz_internal.mz_source_statistics
                Key: (#0{id}, #1{replica_id})
    cte l5 =
      →Arranged mz_catalog.mz_sinks
    cte l6 =
      →Differential Join %0:l5[#0{id}] » %1:mz_sink_statuses[#0{id}]
        →Arranged l5
        →Arranged mz_internal.mz_sink_statuses
    cte l7 =
      →Differential Join %0[#1{cluster_id}] » %1:mz_cluster_replicas[#1{cluster_id}]
        →Arrange (#1{cluster_id})
          →Union
            →Map/Filter/Project
              Project: #0..=#2
              Map: null
                →Consolidating Union
                  →Negate Diffs
                    →Differential Join %0:l5[#0{id}] » %1[#0]
                      →Arranged l5
                      →Distinct GroupAggregate
                        →Fused with Child Map/Filter/Project
                          Project: #0
                            →Read l6
                  →Fused with Child Map/Filter/Project
                    Project: #0, #11
                      →Arranged mz_catalog.mz_sinks
                        Key: (#0{id})
            →Stream l6
        →Arrange (#1{cluster_id})
          →Fused with Child Map/Filter/Project
            Project: #0, #2
              →Arranged mz_catalog.mz_cluster_replicas
                Key: (#0{id})
    cte l8 =
      →Arrange (#0{id}, #2{id})
        →Stream l7
    cte l9 =
      →Differential Join %1:mz_cluster_replica_frontiers[#0{object_id}, #1{replica_id}] » %0:l8[#0{id}, #2{id}]
        →Arranged l8
        →Arranged mz_catalog.mz_cluster_replica_frontiers
  →Return
    →Union
      →Map/Filter/Project
        Project: #0, #1, #3
        Map: coalesce(#2{hydrated}, false)
          →Union
            →Map/Filter/Project
              Project: #0..=#2
              Map: null, null
                →Consolidating Union
                  →Negate Diffs
                    →Unarranged Raw Stream
                      →Distinct GroupAggregate
                        →Fused with Child Map/Filter/Project
                          Project: #0
                            →Read l2
                  →Fused with Child Map/Filter/Project
                    Project: #0
                      →Arranged mz_catalog.mz_indexes
                        Key: (#0{id})
            →Stream l2
      →Map/Filter/Project
        Project: #0, #1, #3
        Map: coalesce(#2{hydrated}, false)
          →Union
            →Map/Filter/Project
              Project: #0..=#2
              Map: null, null
                →Consolidating Union
                  →Negate Diffs
                    →Unarranged Raw Stream
                      →Distinct GroupAggregate
                        →Fused with Child Map/Filter/Project
                          Project: #0
                            →Read l3
                  →Fused with Child Map/Filter/Project
                    Project: #0
                      →Arranged mz_catalog.mz_materialized_views
                        Key: (#0{id})
            →Stream l3
      →Map/Filter/Project
        Project: #0, #1, #3
        Map: (#2{rehydration_latency}) IS NOT NULL
          →Union
            →Map/Filter/Project
              Project: #0..=#2
              Map: null, null
                →Consolidating Union
                  →Negate Diffs
                    →Unarranged Raw Stream
                      →Distinct GroupAggregate
                        →Fused with Child Map/Filter/Project
                          Project: #0
                            →Read l4
                  →Fused with Child Map/Filter/Project
                    Project: #0
                    Filter: (#4{type} != "webhook") AND (#10{cluster_id}) IS NOT NULL
                      →Arranged mz_catalog.mz_sources
                        Key: (#0{id})
            →Stream l4
      →Map/Filter/Project
        Project: #0, #2, #4
        Map: ((#1{status} = "running") AND (coalesce(#3{write_frontier}, 0) > 0))
          →Union
            →Map/Filter/Project
              Project: #0..=#3
              Map: null
                →Consolidating Union
                  →Negate Diffs
                    →Differential Join %1[#0, #1] » %0:l8[#0{id}, #2{id}]
                      →Arranged l8
                      →Distinct GroupAggregate
                        →Fused with Child Map/Filter/Project
                          Project: #0, #2
                            →Read l9
                  →Stream l7
            →Stream l9
Used Indexes:
  - mz_internal.mz_source_statistics_ind (*** full scan ***)
  - mz_catalog.mz_indexes_ind (*** full scan ***, differential join)
  - mz_catalog.mz_sources_ind (*** full scan ***, differential join)
  - mz_catalog.mz_sinks_ind (*** full scan ***, differential join)
  - mz_catalog.mz_materialized_views_ind (*** full scan ***, differential join)
  - mz_internal.mz_sink_statuses_ind (differential join)
  - mz_catalog.mz_cluster_replicas_ind (*** full scan ***)
  - mz_catalog.mz_cluster_replica_frontiers_ind (*** full scan ***, differential join)
  - mz_internal.mz_compute_hydration_times_ind (*** full scan ***)
Target cluster: mz_catalog_server

[antiguru]

mz_hydration_statuses_ind plan diff (the only maintained dataflow that changes)

< before (object_id index) · > after ((object_id, replica_id) index). Three hunks, both in the mz_cluster_replica_frontiers accesses:

  cte l1 = ... Differential Join %0:l0[#0{id}] » %1:mz_cluster_replica_frontiers[#0{object_id}]
-           after %1:                                              # before: post-join filter,
-             Project: #0, #10                                     #   reused the object_id index directly
-             Filter: (#11{write_frontier}) IS NULL
            Final closure: ...
            →Arranged l0
-           →Arranged mz_catalog.mz_cluster_replica_frontiers      # before: index reuse (full)
+           →Arrange (#0{object_id})                               # after: own arrangement, filter pushed
+             →Fused with Child Map/Filter/Project                 #   BEFORE the arrange => small
+               Project: #0, #1
+               Filter: (#2{write_frontier}) IS NULL               #   (write_frontier IS NULL subset)
+                 →Arranged mz_catalog.mz_cluster_replica_frontiers
+                   Key: (#0{object_id}, #1{replica_id})           #   reads the 2-col index

  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})                   # before: FRESH full arrangement (the ~47 GB)
-         →Arranged mz_catalog.mz_cluster_replica_frontiers
+       →Arranged mz_catalog.mz_cluster_replica_frontiers          # after: reuses the index, no fresh arrangement

• l9 (sinks branch): drops its fresh full (object_id, replica_id) arrangement of mz_cluster_replica_frontiers — the arrangement that reached ~47 GB under churn — and reads the index directly (join operand order flips).
• l1 (index/MV branch): instead of reusing the old object_id index with a post-join write_frontier IS NULL filter, it now builds its own object_id arrangement with that filter pushed before the arrange, so it only holds the small NULL-frontier subset.

Net: two full-relation arrangements → one (the index, shared) + one small filtered one. Full before/after plans for both relevant indexes are in the earlier comment.

[ggevay]

We've discussed with @antiguru on zoom, and a thing that could make such changes much easier to evaluate is to have an slt (maybe auto-generated), which would show the EXPLAINs for all the objects on mz_catalog_server (indexes, MVs, and also SELECT * FROM <view>). Then PRs like this would show the plan changes in their diff without relying on Claude doing brittle work to collect plan changes. (For example, it was mentioned on zoom that maybe some mcp view's plan has also changed, but Claude's above comments are not showing that.)

@SangJunBak, this could be considered as a work item in the mz_catalog_server improvements project. I'd be happy to work on this particular bit!

[aljoscha]

Could we use catalog_server: as the commit/pr prefix? We also have an actual catalog component so that can be confusing, please.

[SangJunBak]

@ggevay Sounds like a good idea! Nick has created a ticket here https://linear.app/materializeinc/issue/SQL-351/add-an-slt-maybe-auto-generated-showing-the-explains-for-all-the