store: Create postponed indexes on subgraph restart

Replace the IndexList-based `recreate_invalid_indexes` call in
`start_subgraph()` with a call to `create_postponed_indexes()`. This
uses `IF NOT EXISTS` and `CONCURRENTLY` to safely create any missing
postponed indexes on every restart, acting as a safety net.

Remove the now-unused `IndexList::recreate_invalid_indexes` method.

Author: lutter

store/postgres/src/deployment_store.rs

@@ -1627,19 +1627,15 @@ impl DeploymentStore {
             .await?;
         }
 
-        let mut conn = self.pool.get_permitted().await?;
-        if ENV_VARS.postpone_attribute_index_creation {
-            // Check if all indexes are valid and recreate them if they
-            // aren't.
-            IndexList::load(&mut conn, &dst)
-                .await?
-                .recreate_invalid_indexes(&mut conn, &dst)
-                .await?;
-        }
+        // Create any indexes whose creation was postponed when the
+        // deployment was first created. Using `IF NOT EXISTS` and
+        // `CONCURRENTLY` makes this safe to call on every restart.
+        self.create_postponed_indexes(site.cheap_clone()).await?;
 
         // Make sure the block pointer is set. This is important for newly
         // deployed subgraphs so that we respect the 'startBlock' setting
         // the first time the subgraph is started
+        let mut conn = self.pool.get_permitted().await?;
         conn.transaction(|conn| {
             crate::deployment::initialize_block_ptr(conn, &dst.site).scope_boxed()
         })

store/postgres/src/relational/index.rs

@@ -4,7 +4,6 @@ use std::collections::HashMap;
 use std::fmt::{Display, Write};
 
 use diesel::sql_query;
-use diesel::sql_types::{Bool, Text};
 use diesel_async::RunQueryDsl;
 use graph::components::store::StoreError;
 use graph::itertools::Itertools;
@@ -947,63 +946,6 @@ impl IndexList {
 
         iter
     }
-
-    pub async fn recreate_invalid_indexes(
-        &self,
-        conn: &mut AsyncPgConnection,
-        layout: &Layout,
-    ) -> Result<(), StoreError> {
-        #[derive(QueryableByName, Debug)]
-        struct IndexInfo {
-            #[diesel(sql_type = Bool)]
-            isvalid: bool,
-        }
-
-        let namespace = &layout.catalog.site.namespace;
-        let creat = layout.index_creator(true, true);
-        for table in layout.tables.values() {
-            let idxs = self
-                .indexes_for_table(table)
-                .filter(|idx| idx.to_postpone());
-            for idx in idxs {
-                if let Some(index_name) = idx.name() {
-                    let table_name = table.name.clone();
-                    let query = r#"
-                        SELECT  x.indisvalid           AS isvalid
-                        FROM pg_index x
-                                JOIN pg_class c ON c.oid = x.indrelid
-                                JOIN pg_class i ON i.oid = x.indexrelid
-                                LEFT JOIN pg_namespace n ON n.oid = c.relnamespace
-                        WHERE (c.relkind = ANY (ARRAY ['r'::"char", 'm'::"char", 'p'::"char"]))
-                        AND (i.relkind = ANY (ARRAY ['i'::"char", 'I'::"char"]))
-                        AND (n.nspname = $1)
-                        AND (c.relname = $2)
-                        AND (i.relname = $3);"#;
-                    let ii_vec = sql_query(query)
-                        .bind::<Text, _>(namespace.to_string())
-                        .bind::<Text, _>(table_name)
-                        .bind::<Text, _>(index_name.clone())
-                        .get_results::<IndexInfo>(conn)
-                        .await?
-                        .into_iter()
-                        .collect::<Vec<IndexInfo>>();
-                    assert!(ii_vec.len() <= 1);
-                    if ii_vec.is_empty() || !ii_vec[0].isvalid {
-                        // if a bad index exist lets first drop it
-                        if !ii_vec.is_empty() {
-                            let drop_query =
-                                sql_query(format!("DROP INDEX {}.{};", namespace, index_name));
-                            drop_query.execute(conn).await?;
-                        }
-                        // We are creating concurrently, which can't be done
-                        // in a transaction
-                        IndexCreator::execute(&creat, conn, idx).await?;
-                    }
-                }
-            }
-        }
-        Ok(())
-    }
 }
 
 #[cfg(test)]