store: update namespace handling in table creation and index loading

Author: shiyasmohd

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")),

store/postgres/src/relational/prune.rs

@@ -23,7 +23,47 @@ use crate::{
     vid_batcher::{VidBatcher, VidRange},
 };
 
-use super::{Catalog, Layout, Namespace};
+use super::{
+    index::{load_indexes_from_table, CreateIndex, IndexList},
+    Catalog, Layout, Namespace,
+};
+
+// Additions to `Table` that are useful for pruning
+impl Table {
+    /// Return the first and last vid of any entity that is visible in the
+    /// block range from `first_block` (inclusive) to `last_block`
+    /// (exclusive)
+    fn vid_range(
+        &self,
+        conn: &mut PgConnection,
+        first_block: BlockNumber,
+        last_block: BlockNumber,
+    ) -> Result<(i64, i64), StoreError> {
+        #[derive(QueryableByName)]
+        struct VidRange {
+            #[diesel(sql_type = BigInt)]
+            min_vid: i64,
+            #[diesel(sql_type = BigInt)]
+            max_vid: i64,
+        }
+
+        // Determine the last vid that we need to copy
+        let VidRange { min_vid, max_vid } = sql_query(format!(
+            "/* controller=prune,first={first_block},last={last_block} */ \
+                     select coalesce(min(vid), 0) as min_vid, \
+                            coalesce(max(vid), -1) as max_vid from {src} \
+                      where lower(block_range) <= $2 \
+                        and coalesce(upper(block_range), 2147483647) > $1 \
+                        and coalesce(upper(block_range), 2147483647) <= $2 \
+                        and block_range && int4range($1, $2)",
+            src = self.qualified_name,
+        ))
+        .bind::<Integer, _>(first_block)
+        .bind::<Integer, _>(last_block)
+        .get_result::<VidRange>(conn)?;
+        Ok((min_vid, max_vid))
+    }
+}
 
 /// Utility to copy relevant data out of a source table and into a new
 /// destination table and replace the source table with the destination
@@ -59,7 +99,10 @@ impl TablePair {
             let mut list = IndexList {
                 indexes: HashMap::new(),
             };
-            let indexes = load_indexes_from_table(conn, &src, src_nsp.as_str())?;
+            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,