store: Fix postponed indexes of copies and grafts

store/postgres/src/catalog.rs

@@ -912,31 +912,37 @@ pub(crate) async fn index_creation_is_running(
         .map_err::<StoreError, _>(Into::into)
 }
 
+/// The index definitions of a table, only the valid ones with `valid_only`.
 pub(crate) async fn indexes_for_table(
     conn: &mut AsyncPgConnection,
     schema_name: &str,
     table_name: &str,
+    valid_only: bool,
 ) -> Result<Vec<String>, StoreError> {
     #[derive(Queryable, QueryableByName)]
-    struct IndexName {
+    struct IndexDef {
         #[diesel(sql_type = Text)]
-        #[diesel(column_name = indexdef)]
         def: String,
     }
 
     let query = "
         select
-            indexdef
+            pg_get_indexdef(i.indexrelid) as def
         from
-            pg_indexes
+            pg_index i
+            join pg_class t on t.oid = i.indrelid
+            join pg_class c on c.oid = i.indexrelid
+            join pg_namespace n on n.oid = t.relnamespace
         where
-            schemaname = $1
-            and tablename = $2
-        order by indexname";
+            n.nspname = $1
+            and t.relname = $2
+            and (i.indisvalid or not $3)
+        order by c.relname";
     let results = sql_query(query)
         .bind::<Text, _>(schema_name)
         .bind::<Text, _>(table_name)
-        .load::<IndexName>(conn)
+        .bind::<Bool, _>(valid_only)
+        .load::<IndexDef>(conn)
         .await
         .map_err::<StoreError, _>(Into::into)?;
 

store/postgres/src/copy.rs

@@ -12,7 +12,6 @@
 //! operation can resume after an interruption, for example, because
 //! `graph-node` was restarted while the copy was running.
 use std::{
-    collections::HashSet,
     convert::TryFrom,
     future::Future,
     pin::Pin,
@@ -103,6 +102,14 @@ pub enum Status {
     Cancelled,
 }
 
+/// The source of a copy, as read when the copy starts.
+pub(crate) struct CopySource {
+    /// The source's indexes.
+    pub(crate) list: IndexList,
+    /// Whether the source has built its postponed indexes.
+    pub(crate) postponed_indexes_created: bool,
+}
+
 struct CopyState {
     src: Arc<Layout>,
     dst: Arc<Layout>,
@@ -299,45 +306,39 @@ impl CopyState {
         self.finished.iter().chain(self.unfinished.iter())
     }
 
-    /// Create the indexes that were postponed at the start of the
-    /// copy/graft operation: first the ones that existed in the source
-    /// subgraph, then any indexes for new fields that don't exist in the
-    /// source.
+    /// Create the postponed indexes `dst` needs and give it the source's
+    /// `postponed_indexes_created` flag.
     async fn create_indexes(
         &self,
         conn: &mut AsyncPgConnection,
-        index_list: &IndexList,
+        source: &CopySource,
     ) -> Result<(), StoreError> {
         let creat = self.dst.index_creator(false, true);
-        let dst_nsp = self.dst.site.namespace.to_string();
-
-        // Create the indexes that existed in the source subgraph
-        for table in self.all_tables() {
-            let idxs = index_list
-                .indexes_for_table(&table.dst)
-                .filter(|idx| idx.to_postpone())
-                .map(|idx| idx.with_nsp(dst_nsp.clone()))
-                .collect::<Result<Vec<_>, _>>()?;
 
+        for dst_table in self.dst.tables.values() {
+            // May already hold postponed indexes from an interrupted run or an older version's copy.
+            let existing = dst_table.valid_indexes(conn).await?;
+            let src = self
+                .src
+                .table_for_entity(&dst_table.object)
+                .ok()
+                .map(|src_table| {
+                    let indexes = source
+                        .list
+                        .indexes
+                        .get(src_table.name.as_str())
+                        .map(Vec::as_slice)
+                        .unwrap_or(&[]);
+                    (src_table.as_ref(), indexes)
+                });
+            let idxs =
+                dst_table.postponed_indexes_to_create(&self.dst.input_schema, &existing, src)?;
             creat.execute_many(conn, &idxs).await?;
         }
 
-        // Create indexes for new fields that don't exist in the source
-        for table in self.all_tables() {
-            let src_columns: HashSet<&str> =
-                table.src.columns.iter().map(|c| c.name.as_str()).collect();
-            let new_idxs: Vec<_> = table
-                .dst
-                .indexes(&self.dst.input_schema)
-                .map_err(|_| internal_error!("failed to generate indexes for copy"))?
-                .into_iter()
-                .filter(|idx| idx.to_postpone() && idx.references_column_not_in(&src_columns))
-                .collect();
-            creat.execute_many(conn, &new_idxs).await?;
-        }
-
+        let created = source.postponed_indexes_created;
         conn.transaction(async |conn| {
-            deployment::set_postponed_indexes_created(conn, &self.dst.site).await
+            deployment::set_postponed_indexes_created(conn, &self.dst.site, created).await
         })
         .await?;
 
@@ -1191,7 +1192,7 @@ impl Connection {
         }
     }
 
-    async fn copy_data_internal(&mut self, index_list: IndexList) -> Result<Status, StoreError> {
+    async fn copy_data_internal(&mut self, source: CopySource) -> Result<Status, StoreError> {
         let src = self.src.clone();
         let dst = self.dst.clone();
         let target_block = self.target_block.clone();
@@ -1282,7 +1283,7 @@ impl Connection {
         }
         debug_assert!(self.conn.is_some());
 
-        state.create_indexes(self.get_conn()?, &index_list).await?;
+        state.create_indexes(self.get_conn()?, &source).await?;
 
         self.copy_private_data_sources(&state).await?;
 
@@ -1299,7 +1300,7 @@ impl Connection {
     /// block is guaranteed to not be subject to chain reorgs. All data up
     /// to and including `target_block` will be copied.
     ///
-    /// The parameter index_list is a list of indexes that exist on the `src`.
+    /// The parameter `source` describes the indexes of `src`.
     ///
     /// The copy logic makes heavy use of the fact that the `vid` and
     /// `block_range` of entity versions are related since for two entity
@@ -1309,7 +1310,7 @@ impl Connection {
     /// lower(v1.block_range) => v2.vid > v1.vid` and we can therefore stop
     /// the copying of each table as soon as we hit `max_vid = max { v.vid |
     /// lower(v.block_range) <= target_block.number }`.
-    pub async fn copy_data(mut self, index_list: IndexList) -> Result<Status, StoreError> {
+    pub async fn copy_data(mut self, source: CopySource) -> Result<Status, StoreError> {
         // We require sole access to the destination site, and that we get a
         // consistent view of what has been copied so far. In general, that
         // is always true. It can happen though that this function runs when
@@ -1329,7 +1330,7 @@ impl Connection {
         };
         conn.lock(&self.logger, &dst_site).await?;
 
-        let res = self.copy_data_internal(index_list).await;
+        let res = self.copy_data_internal(source).await;
 
         match self.conn.as_mut() {
             None => {

store/postgres/src/deployment.rs

@@ -854,17 +854,18 @@ pub async fn postponed_indexes_created(
     Ok(created)
 }
 
-/// Mark the postponed (deferred) attribute indexes as created for this
-/// deployment. Once set, we never recreate these indexes — even if an
-/// external process removes them.
+/// Record whether the postponed (deferred) attribute indexes have been
+/// created for this deployment. Once set, we never recreate these indexes —
+/// even if an external process removes them.
 pub async fn set_postponed_indexes_created(
     conn: &mut AsyncPgConnection,
     site: &Site,
+    created: bool,
 ) -> Result<(), StoreError> {
     use deployment as d;
 
     update(d::table.filter(d::id.eq(site.id)))
-        .set(d::postponed_indexes_created.eq(true))
+        .set(d::postponed_indexes_created.eq(created))
         .execute(conn)
         .await?;
     Ok(())

store/postgres/src/deployment_store.rs

@@ -48,6 +48,7 @@ use graph::prelude::{
 use graph::schema::{ApiSchema, EntityKey, EntityType, InputSchema};
 
 use crate::block_range::{BLOCK_COLUMN, BLOCK_RANGE_COLUMN};
+use crate::copy::CopySource;
 use crate::deployment::{self, OnSync};
 use crate::detail::ErrorDetail;
 use crate::dynds::DataSourcesTable;
@@ -711,16 +712,58 @@ impl DeploymentStore {
         let table = resolve_table_name(&layout, &entity_name)?;
         let table_name = &table.name;
         let indexes =
-            catalog::indexes_for_table(&mut conn, schema_name.as_str(), table_name.as_str())
+            catalog::indexes_for_table(&mut conn, schema_name.as_str(), table_name.as_str(), false)
                 .await?;
         Ok(indexes.into_iter().map(CreateIndex::parse).collect())
     }
 
-    pub(crate) async fn load_indexes(&self, site: Arc<Site>) -> Result<IndexList, StoreError> {
+    /// Load `site` as the source of a copy.
+    pub(crate) async fn load_copy_source(&self, site: Arc<Site>) -> Result<CopySource, StoreError> {
         let mut conn = self.pool.get_permitted().await?;
+        // Read the flag before the indexes, so that a set flag never comes
+        // with a list from the middle of the background build.
+        let postponed_indexes_created =
+            deployment::postponed_indexes_created(&mut conn, &site).await?;
         let layout = self.layout(&mut conn, site).await?;
+        let list = IndexList::load(&mut conn, &layout).await?;
+        Ok(CopySource {
+            list,
+            postponed_indexes_created,
+        })
+    }
+
+    pub(crate) async fn postponed_indexes_created(&self, site: &Site) -> Result<bool, StoreError> {
+        let mut conn = self.pool.get_permitted().await?;
+        deployment::postponed_indexes_created(&mut conn, site).await
+    }
 
-        IndexList::load(&mut conn, &layout).await
+    // Only used for tests
+    #[cfg(debug_assertions)]
+    pub(crate) async fn set_postponed_indexes_created(
+        &self,
+        site: &Site,
+        created: bool,
+    ) -> Result<(), StoreError> {
+        let mut conn = self.pool.get_permitted().await?;
+        deployment::set_postponed_indexes_created(&mut conn, site, created).await
+    }
+
+    // Only used for tests
+    #[cfg(debug_assertions)]
+    pub(crate) async fn create_btree_index(
+        &self,
+        site: &Site,
+        name: &str,
+        table: &str,
+        expr: &str,
+    ) -> Result<(), StoreError> {
+        let mut conn = self.pool.get_permitted().await?;
+        let query = format!(
+            "create index \"{}\" on \"{}\".\"{}\" using btree ({})",
+            name, site.namespace, table, expr
+        );
+        sql_query(query).execute(&mut conn).await?;
+        Ok(())
     }
 
     /// Drops an index for a given deployment, concurrently.
@@ -1549,12 +1592,12 @@ impl DeploymentStore {
         &self,
         logger: &Logger,
         site: Arc<Site>,
-        graft_src: Option<(Arc<Layout>, BlockPtr, SubgraphDeploymentEntity, IndexList)>,
+        graft_src: Option<(Arc<Layout>, BlockPtr, SubgraphDeploymentEntity, CopySource)>,
     ) -> Result<(), StoreError> {
         let dst = self.find_layout(site.cheap_clone()).await?;
 
         // If `graft_src` is `Some`, then there is a pending graft.
-        if let Some((src, block, src_deployment, index_list)) = graft_src {
+        if let Some((src, block, src_deployment, source)) = graft_src {
             info!(
                 logger,
                 "Initializing graft by copying data from {} to {}",
@@ -1585,7 +1628,7 @@ impl DeploymentStore {
                 dst_manifest_idx_and_name,
             )
             .await?;
-            let status = copy_conn.copy_data(index_list).await?;
+            let status = copy_conn.copy_data(source).await?;
             if status == crate::copy::Status::Cancelled {
                 return Err(StoreError::Canceled);
             }
@@ -1706,14 +1749,6 @@ impl DeploymentStore {
             catalog::index_creation_is_running(&mut conn, site.namespace.as_str()).await
         }
 
-        async fn postponed_indexes_created(
-            store: &DeploymentStore,
-            site: &Site,
-        ) -> Result<bool, StoreError> {
-            let mut conn = store.pool.get_permitted().await?;
-            deployment::postponed_indexes_created(&mut conn, site).await
-        }
-
         async fn wait_for_concurrent_index_creation(
             logger: &Logger,
             store: &DeploymentStore,
@@ -1787,19 +1822,18 @@ impl DeploymentStore {
             // connection for the whole time. Instead, we get a new
             // connection for each index that we create.
 
-            if postponed_indexes_created(&store, &site).await? {
+            if store.postponed_indexes_created(&site).await? {
                 return Ok(());
             }
 
             for table in layout.tables.values() {
-                let indexes = table.indexes(&layout.input_schema).map_err(|e| {
-                    StoreError::ConstraintViolation(format!("failed to generate indexes: {}", e))
-                })?;
+                let existing = {
+                    let mut conn = store.pool.get_permitted().await?;
+                    table.valid_indexes(&mut conn).await?
+                };
+                let indexes =
+                    table.postponed_indexes_to_create(&layout.input_schema, &existing, None)?;
                 for idx in indexes {
-                    if !idx.to_postpone() {
-                        continue;
-                    }
-
                     wait_for_concurrent_index_creation(&logger, &store, &site).await?;
 
                     create_index(&logger, &store, &layout, &site, &idx).await?;
@@ -1812,7 +1846,7 @@ impl DeploymentStore {
             }
 
             let mut conn = store.pool.get_permitted().await?;
-            deployment::set_postponed_indexes_created(&mut conn, &site).await?;
+            deployment::set_postponed_indexes_created(&mut conn, &site, true).await?;
             Ok(())
         }