wip graft recreation of indexes for changed entity fields

store/postgres/src/copy.rs

@@ -808,8 +808,7 @@ impl Connection {
         }
 
         // Create indexes for all the btree attributes that were postponed at the start of
-        // the copy/graft operations. If they weren't postponed it's still fine to run it
-        // as the creation query checks if they alreadey exist.
+        // the copy/graft operations. First recreate the ones from the original subgraph.
         let conn = self.conn.deref_mut();
         let namespace = self.dst.site.namespace.as_str().to_string();
         for table in state.tables.iter() {
@@ -821,11 +820,23 @@ impl Connection {
                 true,
             );
 
+            println!("POSTPONED: {:?}", arr);
+
             for sql in arr {
                 let query = sql_query(format!("{};", sql));
                 query.execute(conn)?;
             }
         }
+        // Second create the ones for the new fields by skipping those created in the first step.
+        // Here we create both the initial and postponed ones.
+        // TODO: above
+        // let conn = self.conn.deref_mut();
+        // for table in state.tables.iter() {
+        //     for sql in table.batch.dst.create_postponed_indexes().into_iter() {
+        //         let query = sql_query(sql);
+        //         query.execute(conn)?;
+        //     }
+        // }
 
         self.copy_private_data_sources(&state)?;
 

store/postgres/src/deployment_store.rs

@@ -2097,15 +2097,28 @@ impl IndexList {
         let mut arr = vec![];
         if let Some(vec) = self.indexes.get(table_name) {
             for ci in vec {
+                println!("ci: {:?}", ci);
                 if ci.all_colums_in_dest(columns) {
-                    if !ci.is_constraint() && !ci.is_pkey() && postponed == ci.to_postpone() {
-                        if let Ok(sql) = ci
-                            .with_nsp(namespace.clone())
-                            .to_sql(concurent_if_not_exist, concurent_if_not_exist)
-                        {
-                            arr.push(sql)
+                    if !ci.is_constraint() && !ci.is_pkey() {
+                        if postponed == ci.to_postpone() {
+                            if let Ok(sql) = ci
+                                .with_nsp(namespace.clone())
+                                .to_sql(concurent_if_not_exist, concurent_if_not_exist)
+                            {
+                                arr.push(sql)
+                            }
+                        } else {
+                            println!(
+                                "diff stage({}): {}",
+                                postponed,
+                                ci.to_sql(false, false).unwrap()
+                            )
                         }
+                    } else {
+                        println!("skip: {}", ci.to_sql(false, false).unwrap())
                     }
+                } else {
+                    println!("miss in dest: {}", ci.to_sql(false, false).unwrap())
                 }
             }
         }

store/postgres/src/relational/ddl.rs

@@ -374,13 +374,20 @@ impl Table {
         self.create_table(out)?;
         if is_copy_op && ENV_VARS.postpone_attribute_index_creation {
             if let Some(ind_def) = index_def {
+                // TODO: add the rest of colums like block_range etc. and pass as a parameter
+                let mut cols = self
+                    .columns
+                    .iter()
+                    .map(|i| i.name.to_string())
+                    .collect::<Vec<String>>();
                 let arr = ind_def.indexes_for_table(
                     &self.namespace.to_string(),
                     &self.name.to_string(),
                     &self.columns,
                     false,
                     false,
                 );
+                println!("CREATE INITIALY: {:?}", arr);
                 for sql in arr {
                     writeln!(out, "{};", sql).expect("properly formated index statements")
                 }

store/postgres/src/relational/index.rs

@@ -594,42 +594,78 @@ impl CreateIndex {
     }
 
     pub fn all_colums_in_dest(&self, columns: &Vec<Column>) -> bool {
+        fn column_exists(columns: &Vec<String>, column_name: &String) -> bool {
+            for c in columns.iter() {
+                if *c == *column_name {
+                    return true;
+                }
+            }
+            false
+        }
+
+        fn some_column_contained(expresion: &String, columns: &Vec<String>) -> bool {
+            for c in columns.iter() {
+                if expresion.contains(c) {
+                    return true;
+                }
+            }
+            false
+        }
+
+        let cols = &columns
+            .iter()
+            .map(|i| i.name.to_string())
+            .collect::<Vec<String>>();
+
+        print!("all_colums_in_dest() {}", self);
         match self {
             CreateIndex::Unknown { defn: _ } => (),
-            CreateIndex::Parsed { columns: cols, .. } => {
-                for c in cols {
+            CreateIndex::Parsed {
+                columns: parsed_cols,
+                ..
+            } => {
+                for c in parsed_cols {
                     match c {
                         Expr::Column(column_name) => {
                             // TODO: simplify
-                            if !contains(columns, column_name) {
+                            if !column_exists(cols, column_name) {
                                 return false;
                             }
                         }
                         Expr::Prefix(column_name, _) => {
-                            if !contains(columns, column_name) {
+                            if !column_exists(cols, column_name) {
                                 return false;
                             }
                         }
-                        Expr::BlockRange | Expr::BlockRangeLower | Expr::BlockRangeUpper => {
-                            if !contains(columns, &"block_range".to_string()) {
+                        Expr::BlockRange | Expr::BlockRangeLower | Expr::BlockRangeUpper => (),
+                        Expr::Vid => {
+                            if !column_exists(cols, &"vid".to_string()) {
                                 return false;
                             }
                         }
-                        Expr::Vid => {
-                            if !contains(columns, &"vid".to_string()) {
+                        Expr::Block => {
+                            if !column_exists(cols, &"block".to_string()) {
                                 return false;
                             }
                         }
-                        Expr::Block => {
-                            if !contains(columns, &"block".to_string()) {
+                        Expr::Unknown(expression) => {
+                            println!("Unknown: {}", expression);
+                            if !some_column_contained(expression, cols)
+                                && !some_column_contained(
+                                    expression,
+                                    &vec!["block_range".to_string(), "vid".to_string()],
+                                )
+                            {
+                                println!("Not contain: {}", expression);
+                                println!("any of: {:?}", cols);
                                 return false;
                             }
                         }
-                        _ => return false,
                     }
                 }
             }
         }
+        println!("TRUE");
         true
     }
 
@@ -675,15 +711,6 @@ fn has_prefix(s: &str, prefix: &str) -> bool {
     s.starts_with(prefix) || s.ends_with("\"") && s.starts_with(format!("\"{}", prefix).as_str())
 }
 
-fn contains(columns: &Vec<Column>, column_name: &String) -> bool {
-    for c in columns.iter() {
-        if c.name.to_string() == *column_name {
-            return true;
-        }
-    }
-    false
-}
-
 #[test]
 fn parse() {
     use Method::*;