pass list of indexes when pruning a subgraph

store/postgres/src/relational.rs

@@ -1715,7 +1715,7 @@ impl Table {
     pub fn new_like(&self, namespace: &Namespace, name: &SqlName) -> Arc<Table> {
         let other = Table {
             object: self.object.clone(),
-            nsp: self.nsp.clone(),
+            nsp: namespace.clone(),
             name: name.clone(),
             qualified_name: SqlName::qualified_name(namespace, name),
             columns: self.columns.clone(),

store/postgres/src/relational/ddl.rs

@@ -403,13 +403,7 @@ impl Table {
         if index_def.is_some() && ENV_VARS.postpone_attribute_index_creation {
             let arr = index_def
                 .unwrap()
-                .indexes_for_table(
-                    &catalog.site.namespace,
-                    &self.name.to_string(),
-                    &self,
-                    false,
-                    false,
-                )
+                .indexes_for_table(&self.nsp, &self.name.to_string(), &self, false, false)
                 .map_err(|_| fmt::Error)?;
             for (_, sql) in arr {
                 writeln!(out, "{};", sql).expect("properly formated index statements")

store/postgres/src/relational/index.rs

@@ -440,7 +440,7 @@ impl CreateIndex {
         }
     }
 
-    fn with_nsp(&self, nsp2: String) -> Result<Self, Error> {
+    pub fn with_nsp(&self, nsp2: String) -> Result<Self, Error> {
         let s = self.clone();
         match s {
             CreateIndex::Unknown { defn: _ } => Err(anyhow!("Failed to parse the index")),
@@ -734,6 +734,16 @@ pub struct IndexList {
     pub(crate) indexes: HashMap<String, Vec<CreateIndex>>,
 }
 
+pub fn load_indexes_from_table(
+    conn: &mut PgConnection,
+    table: &Arc<Table>,
+    schema_name: &str,
+) -> Result<Vec<CreateIndex>, StoreError> {
+    let table_name = table.name.as_str();
+    let indexes = catalog::indexes_for_table(conn, schema_name, table_name)?;
+    Ok(indexes.into_iter().map(CreateIndex::parse).collect())
+}
+
 impl IndexList {
     pub fn load(
         conn: &mut PgConnection,
@@ -746,10 +756,8 @@ impl IndexList {
         let schema_name = site.namespace.clone();
         let layout = store.layout(conn, site)?;
         for (_, table) in &layout.tables {
-            let table_name = table.name.as_str();
-            let indexes = catalog::indexes_for_table(conn, schema_name.as_str(), table_name)?;
-            let collect: Vec<CreateIndex> = indexes.into_iter().map(CreateIndex::parse).collect();
-            list.indexes.insert(table_name.to_string(), collect);
+            let indexes = load_indexes_from_table(conn, table, schema_name.as_str())?;
+            list.indexes.insert(table.name.to_string(), indexes);
         }
         Ok(list)
     }

store/postgres/src/relational/prune.rs

@@ -1,4 +1,4 @@
-use std::{fmt::Write, sync::Arc};
+use std::{collections::HashMap, fmt::Write, sync::Arc};
 
 use diesel::{
     connection::SimpleConnection,
@@ -23,7 +23,10 @@ use crate::{
     vid_batcher::{VidBatcher, VidRange},
 };
 
-use super::{Catalog, Layout, Namespace};
+use super::{
+    index::{load_indexes_from_table, CreateIndex, IndexList},
+    Catalog, Layout, Namespace,
+};
 
 /// Utility to copy relevant data out of a source table and into a new
 /// destination table and replace the source table with the destination
@@ -56,9 +59,18 @@ impl TablePair {
         if catalog::table_exists(conn, dst_nsp.as_str(), &dst.name)? {
             writeln!(query, "truncate table {};", dst.qualified_name)?;
         } else {
+            let mut list = IndexList {
+                indexes: HashMap::new(),
+            };
+            let indexes = load_indexes_from_table(conn, &src, src_nsp.as_str())?
+                .into_iter()
+                .map(|index| index.with_nsp(dst_nsp.to_string()))
+                .collect::<Result<Vec<CreateIndex>, _>>()?;
+            list.indexes.insert(src.name.to_string(), indexes);
+
             // In case of pruning we don't do delayed creation of indexes,
             // as the asumption is that there is not that much data inserted.
-            dst.as_ddl(schema, catalog, None, &mut query)?;
+            dst.as_ddl(schema, catalog, Some(&list), &mut query)?;
         }
         conn.batch_execute(&query)?;