Aggregate multiprocess replica metrics for console overview view

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

Author: SangJunBak

src/catalog/src/builtin.rs

@@ -8104,7 +8104,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 {
@@ -8142,70 +8143,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
 ),
 /*
@@ -8215,7 +8233,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,
@@ -8241,10 +8259,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

@@ -958,6 +958,13 @@ ArrangeBy[[CallUnary { func: CastStringToVarChar(CastStringToVarChar { length: N
 ArrangeBy[[CallUnary { func: CastStringToVarChar(CastStringToVarChar { length: None, fail_on_len: false }), expr: Column(1) }, CallUnary { func: CastInt64ToString(CastInt64ToString), expr: Column(0) }]]-errors
 ArrangeBy[[Column(0), Column(1)]]
 ArrangeBy[[Column(0), Column(1)]]-errors
+ArrangeBy[[Column(0)]]
+ArrangeBy[[Column(0)]]
+ArrangeBy[[Column(0)]]
+ArrangeBy[[Column(0)]]-errors
+Arranged DistinctBy
+DistinctBy
+DistinctByErrorCheck
 
 statement ok
 DROP TABLE t CASCADE
@@ -980,12 +987,13 @@ WHERE mdod.dataflow_name NOT LIKE '%introspection-subscribe%'
 GROUP BY mdod.name
 ORDER BY mdod.name;
 ----
-AccumulableErrorCheck  9
+AccumulableErrorCheck  11
 Arrange␠ReduceMinsMaxes  3
+Arrange␠bundle␠err  1
 Arrange␠export␠iterative  2
 Arrange␠export␠iterative␠err  2
 Arrange␠recursive␠err  4
-ArrangeAccumulable␠[val:␠empty]  9
+ArrangeAccumulable␠[val:␠empty]  11
 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
@@ -1000,16 +1008,17 @@ ArrangeBy[[Column(0),␠Column(1),␠Column(2),␠Column(3),␠Column(4),␠Colu
 ArrangeBy[[Column(0),␠Column(1),␠Column(2),␠Column(3),␠Column(4),␠Column(5)]]-errors  1
 ArrangeBy[[Column(0),␠Column(1),␠Column(2),␠Column(3),␠Column(4)]]  1
 ArrangeBy[[Column(0),␠Column(1),␠Column(2),␠Column(3),␠Column(4)]]-errors  1
-ArrangeBy[[Column(0),␠Column(1)]]  2
+ArrangeBy[[Column(0),␠Column(1)]]  5
 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)]]  157
 ArrangeBy[[Column(0)]]-errors  44
+ArrangeBy[[Column(1),␠CallVariadic␠{␠func:␠ListIndex,␠exprs:␠[Column(2),␠Literal(Ok(Row{[Int64(1)]}),␠ColumnType␠{␠scalar_type:␠Int64,␠nullable:␠false␠})]␠}]]  2
 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
@@ -1027,14 +1036,14 @@ ArrangeBy[[Column(6)]]  3
 ArrangeBy[[Column(6)]]-errors  2
 ArrangeBy[[Column(9)]]  1
 ArrangeBy[[]]  11
-Arranged␠DistinctBy  47
+Arranged␠DistinctBy  48
 Arranged␠MinsMaxesHierarchical␠input  14
 Arranged␠ReduceInaccumulable  3
 Arranged␠TopK␠input  68
 Distinct␠recursive␠err  4
-DistinctBy  47
-DistinctByErrorCheck  47
-ReduceAccumulable  9
+DistinctBy  48
+DistinctByErrorCheck  48
+ReduceAccumulable  11
 ReduceInaccumulable  3
 ReduceInaccumulable␠Error␠Check  3
 ReduceMinsMaxes  3