Aggregate multiprocess replica metrics for console overview view (#31850)

We use this query in the Console to power the Environment Overview page
and cluster replica graphs for the 14 day time period.

<!--
Describe the contents of the PR briefly but completely.

If you write detailed commit messages, it is acceptable to copy/paste
them
here, or write "see commit messages for details." If there is only one
commit
in the PR, GitHub will have already added its commit message above.
-->

### Motivation

Sibling PR to MaterializeInc/console#3733. Query
should reflect the duplicate code there

Gabor noticed the utilization graphs being broken for clusters with
multiple processes.
https://materializeinc.slack.com/archives/CU7ELJ6E9/p1741596295446089

<!--
Which of the following best describes the motivation behind this PR?

  * This PR fixes a recognized bug.

    [Ensure issue is linked somewhere.]

  * This PR adds a known-desirable feature.

    [Ensure issue is linked somewhere.]

  * This PR fixes a previously unreported bug.

    [Describe the bug in detail, as if you were filing a bug report.]

  * This PR adds a feature that has not yet been specified.

[Write a brief specification for the feature, including justification
for its inclusion in Materialize, as if you were writing the original
     feature specification.]

   * This PR refactors existing code.

[Describe what was wrong with the existing code, if it is not obvious.]
-->

### Tips for reviewer

<!--
Leave some tips for your reviewer, like:

    * The diff is much smaller if viewed with whitespace hidden.
    * [Some function/module/file] deserves extra attention.
* [Some function/module/file] is pure code movement and only needs a
skim.

Delete this section if no tips.
-->

### Checklist

- [x] This PR has adequate test coverage / QA involvement has been duly
considered. ([trigger-ci for additional test/nightly
runs](https://trigger-ci.dev.materialize.com/))

I've tested this manually, but I added a unit test in
MaterializeInc/console#3733

Author: SangJunBak

src/catalog/src/builtin.rs

@@ -8105,7 +8105,8 @@ JOIN root_times r USING (id)",
 
 /**
  * This view is used to display the cluster utilization over 14 days bucketed by 8 hours.
- * It's specifically for the Console's environment overview page to speed up load times
+ * It's specifically for the Console's environment overview page to speed up load times.
+ * This query should be kept in sync with MaterializeInc/console/src/api/materialize/cluster/replicaUtilizationHistory.ts
  */
 pub static MZ_CONSOLE_CLUSTER_UTILIZATION_OVERVIEW: LazyLock<BuiltinView> = LazyLock::new(|| {
     BuiltinView {
@@ -8143,70 +8144,87 @@ pub static MZ_CONSOLE_CLUSTER_UTILIZATION_OVERVIEW: LazyLock<BuiltinView> = Lazy
     cluster_id
   FROM mz_catalog.mz_cluster_replicas
 ),
+replica_metrics_history AS (
+  SELECT
+    m.occurred_at,
+    m.replica_id,
+    r.size,
+    (SUM(m.cpu_nano_cores::float8) / s.cpu_nano_cores) / s.processes AS cpu_percent,
+    (SUM(m.memory_bytes::float8) / s.memory_bytes) / s.processes AS memory_percent,
+    (SUM(m.disk_bytes::float8) / s.disk_bytes) / s.processes AS disk_percent,
+    SUM(m.disk_bytes::float8) AS disk_bytes,
+    SUM(m.memory_bytes::float8) AS memory_bytes,
+    s.disk_bytes::numeric * s.processes AS total_disk_bytes,
+    s.memory_bytes::numeric * s.processes AS total_memory_bytes
+  FROM
+    replica_history AS r
+    INNER JOIN mz_catalog.mz_cluster_replica_sizes AS s ON r.size = s.size
+    INNER JOIN mz_internal.mz_cluster_replica_metrics_history AS m ON m.replica_id = r.replica_id
+  GROUP BY
+    m.occurred_at,
+    m.replica_id,
+    r.size,
+    s.cpu_nano_cores,
+    s.memory_bytes,
+    s.disk_bytes,
+    s.processes
+),
 replica_utilization_history_binned AS (
   SELECT m.occurred_at,
     m.replica_id,
-    m.process_id,
-    (m.cpu_nano_cores::float8 / s.cpu_nano_cores) AS cpu_percent,
-    (m.memory_bytes::float8 / s.memory_bytes) AS memory_percent,
-    (m.disk_bytes::float8 / s.disk_bytes) AS disk_percent,
-    m.disk_bytes::float8 AS disk_bytes,
-    m.memory_bytes::float8 AS memory_bytes,
-    s.disk_bytes AS total_disk_bytes,
-    s.memory_bytes AS total_memory_bytes,
-    r.size,
+    m.cpu_percent,
+    m.memory_percent,
+    m.memory_bytes,
+    m.disk_percent,
+    m.disk_bytes,
+    m.total_disk_bytes,
+    m.total_memory_bytes,
+    m.size,
     date_bin(
       '8 HOURS',
       occurred_at,
       '1970-01-01'::timestamp
     ) AS bucket_start
   FROM replica_history AS r
-    JOIN mz_catalog.mz_cluster_replica_sizes AS s ON r.size = s.size
-    JOIN mz_internal.mz_cluster_replica_metrics_history AS m ON m.replica_id = r.replica_id
+    JOIN replica_metrics_history AS m ON m.replica_id = r.replica_id
   WHERE mz_now() <= date_bin(
       '8 HOURS',
       occurred_at,
       '1970-01-01'::timestamp
     ) + INTERVAL '14 DAYS'
 ),
--- For each (replica, process_id, bucket), take the (replica, process_id, bucket) with the highest memory
+-- For each (replica, bucket), take the (replica, bucket) with the highest memory
 max_memory AS (
-  SELECT DISTINCT ON (bucket_start, replica_id, process_id) bucket_start,
+  SELECT DISTINCT ON (bucket_start, replica_id) bucket_start,
     replica_id,
-    process_id,
     memory_percent,
     occurred_at
   FROM replica_utilization_history_binned
   OPTIONS (DISTINCT ON INPUT GROUP SIZE = 480)
   ORDER BY bucket_start,
     replica_id,
-    process_id,
     COALESCE(memory_bytes, 0) DESC
 ),
 max_disk AS (
-  SELECT DISTINCT ON (bucket_start, replica_id, process_id) bucket_start,
+  SELECT DISTINCT ON (bucket_start, replica_id) bucket_start,
     replica_id,
-    process_id,
     disk_percent,
     occurred_at
   FROM replica_utilization_history_binned
   OPTIONS (DISTINCT ON INPUT GROUP SIZE = 480)
   ORDER BY bucket_start,
     replica_id,
-    process_id,
     COALESCE(disk_bytes, 0) DESC
 ),
 max_cpu AS (
-  SELECT DISTINCT ON (bucket_start, replica_id, process_id) bucket_start,
+  SELECT DISTINCT ON (bucket_start, replica_id) bucket_start,
     replica_id,
-    process_id,
     cpu_percent,
     occurred_at
   FROM replica_utilization_history_binned
   OPTIONS (DISTINCT ON INPUT GROUP SIZE = 480)
   ORDER BY bucket_start,
     replica_id,
-    process_id,
     COALESCE(cpu_percent, 0) DESC
 ),
 /*
@@ -8216,7 +8234,7 @@ max_cpu AS (
  values may not occur at the same time if the bucket interval is large.
  */
 max_memory_and_disk AS (
-  SELECT DISTINCT ON (bucket_start, replica_id, process_id) bucket_start,
+  SELECT DISTINCT ON (bucket_start, replica_id) bucket_start,
     replica_id,
     memory_percent,
     disk_percent,
@@ -8242,10 +8260,9 @@ max_memory_and_disk AS (
   OPTIONS (DISTINCT ON INPUT GROUP SIZE = 480)
   ORDER BY bucket_start,
     replica_id,
-    process_id,
     COALESCE(memory_and_disk_percent, 0) DESC
 ),
--- For each (replica, process_id, bucket), get its offline events at that time
+-- For each (replica, bucket), get its offline events at that time
 replica_offline_event_history AS (
   SELECT date_bin(
       '8 HOURS',

test/sqllogictest/distinct_arrangements.slt

@@ -980,12 +980,12 @@ WHERE mdod.dataflow_name NOT LIKE '%introspection-subscribe%'
 GROUP BY mdod.name
 ORDER BY mdod.name;
 ----
-AccumulableErrorCheck  9
+AccumulableErrorCheck  10
 Arrange␠ReduceMinsMaxes  3
 Arrange␠export␠iterative  2
 Arrange␠export␠iterative␠err  2
 Arrange␠recursive␠err  4
-ArrangeAccumulable␠[val:␠empty]  9
+ArrangeAccumulable␠[val:␠empty]  10
 ArrangeBy[[CallBinary␠{␠func:␠JsonbGetString␠{␠stringify:␠true␠},␠expr1:␠Column(1),␠expr2:␠Literal(Ok(Row{[String("id")]}),␠ColumnType␠{␠scalar_type:␠String,␠nullable:␠false␠})␠}]]  2
 ArrangeBy[[CallBinary␠{␠func:␠JsonbGetString␠{␠stringify:␠true␠},␠expr1:␠Column(2),␠expr2:␠Literal(Ok(Row{[String("id")]}),␠ColumnType␠{␠scalar_type:␠String,␠nullable:␠false␠})␠}]]  1
 ArrangeBy[[CallVariadic␠{␠func:␠Coalesce,␠exprs:␠[Column(2),␠Column(3)]␠}]]  2
@@ -1004,12 +1004,12 @@ ArrangeBy[[Column(0),␠Column(1)]]  2
 ArrangeBy[[Column(0),␠Column(2)]]  4
 ArrangeBy[[Column(0),␠Column(3)]]  4
 ArrangeBy[[Column(0),␠Column(4)]]  1
-ArrangeBy[[Column(0)]]  154
+ArrangeBy[[Column(0)]]  155
 ArrangeBy[[Column(0)]]-errors  44
 ArrangeBy[[Column(1),␠Column(0)]]  1
 ArrangeBy[[Column(1),␠Column(2)]]  2
 ArrangeBy[[Column(1),␠Column(3)]]  1
-ArrangeBy[[Column(1)]]  25
+ArrangeBy[[Column(1)]]  26
 ArrangeBy[[Column(1)]]-errors  7
 ArrangeBy[[Column(13)]]  1
 ArrangeBy[[Column(15)]]  1
@@ -1034,7 +1034,7 @@ Arranged␠TopK␠input  68
 Distinct␠recursive␠err  4
 DistinctBy  47
 DistinctByErrorCheck  47
-ReduceAccumulable  9
+ReduceAccumulable  10
 ReduceInaccumulable  3
 ReduceInaccumulable␠Error␠Check  3
 ReduceMinsMaxes  3