Feat: Expose subgraph table sizes in SubgraphIndexingStatus

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/resolver.rs

@@ -123,7 +123,7 @@ impl<S: Store> IndexNodeResolver<S> {
     }
 
     fn resolve_indexing_statuses(&self, field: &a::Field) -> Result<r::Value, QueryExecutionError> {
-        let deployments = field
+        let deployments: Vec<String> = field
             .argument_value("subgraphs")
             .map(|value| match value {
                 r::Value::List(ids) => ids
@@ -135,7 +135,7 @@ impl<S: Store> IndexNodeResolver<S> {
                     .collect(),
                 _ => unreachable!(),
             })
-            .unwrap_or_else(Vec::new);
+            .unwrap();
 
         if deployments.is_empty() {
             return Ok(r::Value::List(vec![]));
@@ -171,6 +171,35 @@ impl<S: Store> IndexNodeResolver<S> {
         Ok(infos.into_value())
     }
 
+    fn resolve_indexing_status_for_version(
+        &self,
+        field: &a::Field,
+
+        // If `true` return the current version, if `false` return the pending version.
+        current_version: bool,
+    ) -> Result<r::Value, QueryExecutionError> {
+        // We can safely unwrap because the argument is non-nullable and has been validated.
+        let subgraph_name = field.get_required::<String>("subgraphName").unwrap();
+
+        debug!(
+            self.logger,
+            "Resolve indexing status for subgraph name";
+            "name" => &subgraph_name,
+            "current_version" => current_version,
+        );
+
+        let infos = self.store.status(status::Filter::SubgraphVersion(
+            subgraph_name,
+            current_version,
+        ))?;
+
+        Ok(infos
+            .into_iter()
+            .next()
+            .map(|info| info.into_value())
+            .unwrap_or(r::Value::Null))
+    }
+
     fn resolve_entity_changes_in_block(
         &self,
         field: &a::Field,
@@ -443,35 +472,6 @@ impl<S: Store> IndexNodeResolver<S> {
         Ok(r::Value::List(public_poi_results))
     }
 
-    fn resolve_indexing_status_for_version(
-        &self,
-        field: &a::Field,
-
-        // If `true` return the current version, if `false` return the pending version.
-        current_version: bool,
-    ) -> Result<r::Value, QueryExecutionError> {
-        // We can safely unwrap because the argument is non-nullable and has been validated.
-        let subgraph_name = field.get_required::<String>("subgraphName").unwrap();
-
-        debug!(
-            self.logger,
-            "Resolve indexing status for subgraph name";
-            "name" => &subgraph_name,
-            "current_version" => current_version,
-        );
-
-        let infos = self.store.status(status::Filter::SubgraphVersion(
-            subgraph_name,
-            current_version,
-        ))?;
-
-        Ok(infos
-            .into_iter()
-            .next()
-            .map(|info| info.into_value())
-            .unwrap_or(r::Value::Null))
-    }
-
     async fn validate_and_extract_features<C, SgStore>(
         subgraph_store: &Arc<SgStore>,
         unvalidated_subgraph_manifest: UnvalidatedSubgraphManifest<C>,

server/index-node/src/schema.graphql

@@ -21,7 +21,7 @@ type Query {
   indexingStatusesForSubgraphName(
     subgraphName: String!
   ): [SubgraphIndexingStatus!]!
-  indexingStatuses(subgraphs: [String!]): [SubgraphIndexingStatus!]!
+  indexingStatuses(subgraphs: [String!]!): [SubgraphIndexingStatus!]!
   proofOfIndexing(
     subgraph: String!
     blockNumber: Int!
@@ -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, PgConnection,
     QueryDsl, RunQueryDsl, SelectableHelper as _,
 };
+use diesel::sql_types::{Array, BigInt, Integer};
 use diesel_derives::Associations;
 use git_testament::{git_testament, git_testament_macros};
 use graph::blockchain::BlockHash;
@@ -239,6 +240,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,
@@ -301,6 +303,7 @@ pub(crate) fn info_from_details(
         entity_count,
         node: None,
         history_blocks: subgraph_history_blocks,
+        subgraph_size,
     })
 }
 
@@ -422,17 +425,93 @@ pub(crate) 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 result = 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 rows = match result {
+        Ok(rows) => rows,
+        Err(e) if e.to_string().contains("has not been populated") => Vec::new(),
+        Err(e) => return Err(e.into()),
+    };
+
+    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))]