graph, store: Mirror source indexes on copy/graft

docs/environment-variables.md

@@ -298,6 +298,13 @@ those.
 - `GRAPH_POSTPONE_ATTRIBUTE_INDEX_CREATION`: During the coping of a subgraph
   postponing creation of certain indexes (btree, attribute based ones), would
   speed up syncing
+- `GRAPH_STORE_COPY_MIRROR_INDEXES`: Whether copies and grafts take the
+  indexes created with their tables from their source: attribute indexes
+  dropped on the source stay dropped, and the source's other indexes,
+  including manual ones, are carried over. When `false`, these indexes
+  come from the destination's own layout, as before. Set it on the node
+  that runs `graphman copy create` and on the nodes that handle deployments
+  (for grafts). Defaults to `true`.
 - `GRAPH_STORE_INSERT_EXTRA_COLS`: Makes it possible to work around bugs in
   the subgraph writing code that manifest as Postgres errors saying 'number
   of parameters must be between 0 and 65535' Such errors are always

graph/src/env/store.rs

@@ -141,6 +141,10 @@ pub struct EnvVarsStore {
     /// Whether to create GIN indexes for array attributes. Set by
     /// `GRAPH_STORE_CREATE_GIN_INDEXES`. The default is `false`
     pub create_gin_indexes: bool,
+    /// Whether copies and grafts take the indexes created with their tables
+    /// from their source instead of their own layout. Set by
+    /// `GRAPH_STORE_COPY_MIRROR_INDEXES`. The default is `true`.
+    pub copy_mirror_indexes: bool,
     /// Temporary env var in case we need to quickly rollback PR #5010
     pub use_brin_for_all_query_types: bool,
     /// Temporary env var to disable certain lookups in the chain store
@@ -239,6 +243,7 @@ impl TryFrom<InnerStore> for EnvVarsStore {
             write_batch_duration: Duration::from_secs(x.write_batch_duration_in_secs),
             write_batch_size: x.write_batch_size * 1_000,
             create_gin_indexes: x.create_gin_indexes,
+            copy_mirror_indexes: x.copy_mirror_indexes,
             use_brin_for_all_query_types: x.use_brin_for_all_query_types,
             disable_block_cache_for_lookup: x.disable_block_cache_for_lookup,
             ignore_block_cache: x.ignore_block_cache,
@@ -346,6 +351,8 @@ pub struct InnerStore {
     write_batch_memoize: bool,
     #[envconfig(from = "GRAPH_STORE_CREATE_GIN_INDEXES", default = "false")]
     create_gin_indexes: bool,
+    #[envconfig(from = "GRAPH_STORE_COPY_MIRROR_INDEXES", default = "true")]
+    copy_mirror_indexes: bool,
     #[envconfig(from = "GRAPH_STORE_USE_BRIN_FOR_ALL_QUERY_TYPES", default = "false")]
     use_brin_for_all_query_types: bool,
     #[envconfig(from = "GRAPH_STORE_DISABLE_BLOCK_CACHE_FOR_LOOKUP", default = "false")]

store/postgres/examples/layout.rs

@@ -64,7 +64,7 @@ fn print_delete_all(layout: &Layout) {
 }
 
 fn print_ddl(layout: &Layout) {
-    let ddl = ensure(layout.as_ddl(), "Failed to generate DDL");
+    let ddl = ensure(layout.as_ddl(None), "Failed to generate DDL");
     println!("{}", ddl);
 }
 

store/postgres/src/copy.rs

@@ -32,7 +32,10 @@ use crate::{
     dynds::DataSourcesTable,
     pool::TxnCallback,
     primary::{DeploymentId, Primary, Site},
-    relational::{Layout, Table, index::IndexList},
+    relational::{
+        Layout, Table,
+        index::{IndexList, mirrored_table},
+    },
     relational_queries as rq,
     vid_batcher::{VidBatcher, VidRange},
 };
@@ -318,19 +321,7 @@ impl CopyState {
         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 src = mirrored_table(dst_table, (self.src.as_ref(), &source.list));
             let idxs =
                 dst_table.postponed_indexes_to_create(&self.dst.input_schema, &existing, src)?;
             creat.execute_many(conn, &idxs).await?;

store/postgres/src/deployment_store.rs

@@ -53,7 +53,7 @@ use crate::deployment::{self, OnSync};
 use crate::detail::ErrorDetail;
 use crate::dynds::DataSourcesTable;
 use crate::primary::{DeploymentId, Primary};
-use crate::relational::index::{CreateIndex, IndexCreator, IndexList, Method};
+use crate::relational::index::{CreateIndex, IndexCreator, IndexList, Method, SourceIndex};
 use crate::relational::{self, Layout, LayoutCache, STATEMENT_TIMEOUT, SqlName, Table};
 use crate::relational_queries::{FromEntityData, JSONData};
 use crate::{AsyncPgConnection, advisory_lock, catalog, retry};
@@ -177,14 +177,18 @@ impl DeploymentStore {
         DeploymentStore(Arc::new(store))
     }
 
+    /// For a copy or graft that mirrors its source, `src` is the source's
+    /// layout and indexes.
     pub(crate) async fn create_deployment(
         &self,
         schema: &InputSchema,
         deployment: DeploymentCreate,
         site: Arc<Site>,
         replace: bool,
         on_sync: OnSync,
+        src: Option<(&Layout, &IndexList)>,
     ) -> Result<(), StoreError> {
+        let logger = self.logger.cheap_clone();
         let mut conn = self.pool.get_permitted().await?;
         conn.transaction::<_, StoreError, _>(async |conn| {
             let exists = deployment::exists(conn, &site).await?;
@@ -211,13 +215,17 @@ impl DeploymentStore {
                 let query = format!("create schema {}", &site.namespace);
                 conn.batch_execute(&query).await?;
 
-                let _ = Layout::create_relational_schema(
+                let layout = Layout::create_relational_schema(
                     conn,
                     site.clone(),
                     schema,
                     entities_with_causality_region.into_iter().collect(),
+                    src,
                 )
                 .await?;
+                if let Some(src) = src {
+                    copy_source_indexes(&logger, conn, &layout, src).await?;
+                }
 
                 // Create data sources table
                 if site.schema_version.private_data_sources() {
@@ -725,7 +733,9 @@ impl DeploymentStore {
         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?;
+        // Invalid indexes are leftovers of interrupted concurrent drops or
+        // builds.
+        let list = IndexList::load(&mut conn, &layout, true).await?;
         Ok(CopySource {
             list,
             postponed_indexes_created,
@@ -756,11 +766,13 @@ impl DeploymentStore {
         name: &str,
         table: &str,
         expr: &str,
+        unique: bool,
     ) -> Result<(), StoreError> {
         let mut conn = self.pool.get_permitted().await?;
+        let unique = if unique { "unique " } else { "" };
         let query = format!(
-            "create index \"{}\" on \"{}\".\"{}\" using btree ({})",
-            name, site.namespace, table, expr
+            "create {}index concurrently \"{}\" on \"{}\".\"{}\" using btree ({})",
+            unique, name, site.namespace, table, expr
         );
         sql_query(query).execute(&mut conn).await?;
         Ok(())
@@ -930,7 +942,7 @@ impl DeploymentStore {
         // Layout and index list are schema metadata — safe to load outside
         // the snapshot transaction
         let layout = self.layout(&mut conn, site.cheap_clone()).await?;
-        let index_list = IndexList::load(&mut conn, &layout).await?;
+        let index_list = IndexList::load(&mut conn, &layout, false).await?;
 
         // Use REPEATABLE READ to get a consistent MVCC snapshot for the
         // entire dump. All queries inside see the same database state,
@@ -2117,6 +2129,37 @@ impl DeploymentStore {
     }
 }
 
+/// Create the indexes of the copy source `src` that `layout` doesn't create
+/// itself. Each runs in its own savepoint, so one that fails is skipped.
+async fn copy_source_indexes(
+    logger: &Logger,
+    conn: &mut AsyncPgConnection,
+    layout: &Layout,
+    src: (&Layout, &IndexList),
+) -> Result<(), StoreError> {
+    for (defn, action) in layout.source_indexes(src)? {
+        match action {
+            SourceIndex::Copy(sql) => {
+                let res = conn
+                    .transaction::<_, StoreError, _>(async |conn| {
+                        sql_query(&sql).execute(conn).await?;
+                        Ok(())
+                    })
+                    .await;
+                if let Err(e) = res {
+                    warn!(logger, "Could not copy an index of the source";
+                        "index" => &defn, "error" => e.to_string());
+                }
+            }
+            SourceIndex::Skip(reason) => {
+                warn!(logger, "Not copying an index of the source";
+                    "index" => &defn, "reason" => reason);
+            }
+        }
+    }
+    Ok(())
+}
+
 /// Tries to fetch a [`Table`] either by its Entity name or its SQL name.
 ///
 /// Since we allow our input to be either camel-case or snake-case, we must retry the

store/postgres/src/relational.rs

@@ -381,12 +381,13 @@ impl Layout {
         site: Arc<Site>,
         schema: &InputSchema,
         entities_with_causality_region: BTreeSet<EntityType>,
+        src: Option<(&Layout, &index::IndexList)>,
     ) -> Result<Layout, StoreError> {
         let catalog =
             Catalog::for_creation(conn, site.cheap_clone(), entities_with_causality_region).await?;
         let layout = Self::new(site, schema, catalog)?;
         let sql = layout
-            .as_ddl()
+            .as_ddl(src)
             .map_err(|_| StoreError::Unknown(anyhow!("failed to generate DDL for layout")))?;
         conn.batch_execute(&sql).await?;
         Ok(layout)

store/postgres/src/relational/ddl.rs

@@ -13,7 +13,9 @@ use crate::{block_range::CAUSALITY_REGION_COLUMN, relational::index::Cond};
 
 use super::{
     Column, Layout, SqlName, Table,
-    index::{CreateIndex, Expr, IndexCreator, Method, PrefixKind},
+    index::{
+        CreateIndex, Expr, IndexCreator, IndexList, Method, PrefixKind, SourceTable, mirrored_table,
+    },
 };
 
 // In debug builds (for testing etc.) unconditionally create exclusion constraints, in release
@@ -28,8 +30,9 @@ impl Layout {
     /// and `create index` etc. statements needed in the database schema
     ///
     /// See the unit tests at the end of this file for the actual DDL that
-    /// gets generated
-    pub fn as_ddl(&self) -> Result<String, fmt::Error> {
+    /// gets generated. For a copy or graft, `src` is the source's layout and
+    /// indexes.
+    pub fn as_ddl(&self, src: Option<(&Layout, &IndexList)>) -> Result<String, fmt::Error> {
         let mut out = String::new();
 
         // Output enums first so table definitions can reference them
@@ -42,7 +45,8 @@ impl Layout {
         // Output 'create table' statements for all tables
         let creat = self.index_creator(false, false);
         for table in tables {
-            table.as_ddl(&self.input_schema, &creat, &mut out)?;
+            let src = src.and_then(|src| mirrored_table(table, src));
+            table.as_ddl(&self.input_schema, &creat, src, &mut out)?;
         }
 
         Ok(out)
@@ -406,19 +410,23 @@ impl Table {
     }
 
     /// Generate the DDL for one table, i.e. one `create table` statement
-    /// and all `create index` statements for the table's columns
+    /// and all `create index` statements for the table's columns. With
+    /// `src`, the table of a copy source, attribute indexes the source
+    /// dropped are left out.
     ///
     /// See the unit tests at the end of this file for the actual DDL that
     /// gets generated
     pub(crate) fn as_ddl(
         &self,
         schema: &InputSchema,
         creat: &IndexCreator,
+        src: Option<SourceTable>,
         out: &mut String,
     ) -> fmt::Result {
         self.create_table(out)?;
         for idx in self.indexes(schema)? {
-            if !idx.to_postpone() {
+            let dropped = src.is_some_and(|src| idx.is_attr_index() && src.dropped(&idx));
+            if !idx.to_postpone() && !dropped {
                 writeln!(out, "{};", creat.to_sql(&idx)?)?;
             }
         }