graph, store, server: Expose table sizes in deployment indexing status

Signed-off-by: Maksim Dimitrov <dimitrov.maksim@gmail.com>

Author: dimitrovmaksim

graph/src/data/subgraph/status.rs

@@ -91,6 +91,35 @@ impl IntoValue for ChainInfo {
     }
 }
 
+#[derive(Debug, Default)]
+pub struct SubgraphSize {
+    pub row_estimate: i64,
+    pub table_bytes: i64,
+    pub index_bytes: i64,
+    pub toast_bytes: i64,
+    pub total_bytes: i64,
+}
+
+impl IntoValue for SubgraphSize {
+    fn into_value(self) -> r::Value {
+        let SubgraphSize {
+            row_estimate,
+            table_bytes,
+            index_bytes,
+            toast_bytes,
+            total_bytes,
+        } = self;
+        object! {
+            __typename: "SubgraphSize",
+            rowEstimate: format!("{}", row_estimate),
+            tableSize: format!("{}", table_bytes),
+            indexSize: format!("{}", index_bytes),
+            toastSize: format!("{}", toast_bytes),
+            totalSize: format!("{}", total_bytes),
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct Info {
     pub id: DeploymentId,
@@ -114,6 +143,9 @@ pub struct Info {
     pub node: Option<String>,
 
     pub history_blocks: i32,
+
+    /// The size of all entity tables in a deployment's namespace in bytes.
+    pub subgraph_size: SubgraphSize,
 }
 
 impl IntoValue for Info {
@@ -130,6 +162,7 @@ impl IntoValue for Info {
             non_fatal_errors,
             synced,
             history_blocks,
+            subgraph_size,
         } = self;
 
         fn subgraph_error_to_value(subgraph_error: SubgraphError) -> r::Value {
@@ -173,6 +206,7 @@ impl IntoValue for Info {
             entityCount: format!("{}", entity_count),
             node: node,
             historyBlocks: history_blocks,
+            subgraphSize: subgraph_size.into_value(),
         }
     }
 }

server/index-node/src/schema.graphql

@@ -77,6 +77,9 @@ type SubgraphIndexingStatus {
   paused: Boolean
 
   historyBlocks: Int!
+
+  "Total size of all tables a deployment's namespace in bytes."
+  subgraphSize: SubgraphSize!
 }
 
 interface ChainIndexingStatus {
@@ -206,3 +209,11 @@ type ApiVersion {
   """
   version: String!
 }
+
+type SubgraphSize {
+  rowEstimate: BigInt!
+  tableSize: BigInt!
+  indexSize: BigInt!
+  toastSize: BigInt!
+  totalSize: BigInt!
+}

store/postgres/src/detail.rs

@@ -7,6 +7,7 @@ use diesel::prelude::{
     ExpressionMethods, JoinOnDsl, NullableExpressionMethods, OptionalExtension, QueryDsl,
     SelectableHelper as _,
 };
+use diesel::sql_types::{Array, BigInt, Integer};
 use diesel_async::RunQueryDsl;
 use diesel_derives::Associations;
 use git_testament::{git_testament, git_testament_macros};
@@ -242,6 +243,7 @@ pub(crate) fn info_from_details(
     non_fatal: Vec<ErrorDetail>,
     sites: &[Arc<Site>],
     subgraph_history_blocks: i32,
+    subgraph_size: status::SubgraphSize,
 ) -> Result<status::Info, StoreError> {
     let DeploymentDetail {
         id,
@@ -304,6 +306,7 @@ pub(crate) fn info_from_details(
         entity_count,
         node: None,
         history_blocks: subgraph_history_blocks,
+        subgraph_size,
     })
 }
 
@@ -434,17 +437,87 @@ pub(crate) async fn deployment_statuses(
         .collect()
     };
 
+    let mut deployment_sizes = deployment_sizes(conn, sites)?;
+
     details_with_fatal_error
         .into_iter()
         .map(|(deployment, head, fatal)| {
             let detail = DeploymentDetail::from((deployment, head));
             let non_fatal = non_fatal_errors.remove(&detail.id).unwrap_or_default();
             let subgraph_history_blocks = history_blocks_map.remove(&detail.id).unwrap_or_default();
-            info_from_details(detail, fatal, non_fatal, sites, subgraph_history_blocks)
+            let table_sizes = deployment_sizes.remove(&detail.id).unwrap_or_default();
+            info_from_details(
+                detail,
+                fatal,
+                non_fatal,
+                sites,
+                subgraph_history_blocks,
+                table_sizes,
+            )
         })
         .collect()
 }
 
+fn deployment_sizes(
+    conn: &mut PgConnection,
+    sites: &[Arc<Site>],
+) -> Result<HashMap<DeploymentId, status::SubgraphSize>, StoreError> {
+    #[derive(QueryableByName)]
+    struct SubgraphSizeRow {
+        #[diesel(sql_type = Integer)]
+        id: DeploymentId,
+        #[diesel(sql_type = BigInt)]
+        row_estimate: i64,
+        #[diesel(sql_type = BigInt)]
+        table_bytes: i64,
+        #[diesel(sql_type = BigInt)]
+        index_bytes: i64,
+        #[diesel(sql_type = BigInt)]
+        toast_bytes: i64,
+        #[diesel(sql_type = BigInt)]
+        total_bytes: i64,
+    }
+
+    let mut query = String::from(
+        r#"
+        SELECT
+            ds.id,
+            ss.row_estimate::bigint,
+            ss.table_bytes::bigint,
+            ss.index_bytes::bigint,
+            ss.toast_bytes::bigint,
+            ss.total_bytes::bigint
+        FROM deployment_schemas ds
+        JOIN info.subgraph_sizes as ss on ss.name = ds.name
+    "#,
+    );
+
+    let rows = if sites.is_empty() {
+        diesel::sql_query(query).load::<SubgraphSizeRow>(conn)?
+    } else {
+        query.push_str(" WHERE ds.id = ANY($1)");
+        diesel::sql_query(query)
+            .bind::<Array<Integer>, _>(sites.iter().map(|site| site.id).collect::<Vec<_>>())
+            .load::<SubgraphSizeRow>(conn)?
+    };
+
+    let mut sizes: HashMap<DeploymentId, status::SubgraphSize> = HashMap::new();
+    for row in rows {
+        sizes.insert(
+            row.id,
+            status::SubgraphSize {
+                row_estimate: row.row_estimate,
+                table_bytes: row.table_bytes,
+                index_bytes: row.index_bytes,
+                toast_bytes: row.toast_bytes,
+                total_bytes: row.total_bytes,
+            },
+        );
+    }
+
+    Ok(sizes)
+}
+
 #[derive(Queryable, Selectable, Identifiable, Associations)]
 #[diesel(table_name = subgraph_manifest)]
 #[diesel(belongs_to(GraphNodeVersion))]