store: Do not emit an index in block$ twice

If an immutable entity type has a `block` column, we would emit the index
on the block$ twice, making deployment fail

Author: lutter

store/postgres/src/relational/ddl_tests.rs

@@ -352,6 +352,74 @@ fn can_copy_from() {
     );
 }
 
+/// Check that we do not create the index on `block$` twice. There was a bug
+/// that if an immutable entity type had a `block` field and index creation
+/// was postponed, we would emit the index on `block$` twice, once from
+/// `Table.create_time_travel_indexes` and once through
+/// `IndexList.indexes_for_table`
+#[test]
+fn postponed_indexes_with_block_column() {
+    fn index_list() -> IndexList {
+        // To generate this list, print the output of `layout.as_ddl(None)`, run
+        // that in Postgres and do `select indexdef from pg_indexes where
+        // schemaname = 'sgd0815'`
+        const INDEX_DEFS: &[&str] = &[
+            "CREATE UNIQUE INDEX data_pkey ON sgd0815.data USING btree (vid)",
+            "CREATE UNIQUE INDEX data_id_key ON sgd0815.data USING btree (id)",
+            "CREATE INDEX data_block ON sgd0815.data USING btree (block$)",
+            "CREATE INDEX attr_1_0_data_block ON sgd0815.data USING btree (block, \"block$\")",
+        ];
+
+        let mut indexes: HashMap<String, Vec<CreateIndex>> = HashMap::new();
+        indexes.insert(
+            "data".to_string(),
+            INDEX_DEFS
+                .iter()
+                .map(|def| CreateIndex::parse(def.to_string()))
+                .collect(),
+        );
+        IndexList { indexes }
+    }
+    // Names of the two indexes we are interested in. Not the leading space
+    // to guard a little against overlapping names
+    const BLOCK_IDX: &str = " data_block";
+    const ATTR_IDX: &str = " attr_1_0_data_block";
+
+    let layout = test_layout(BLOCK_GQL);
+
+    // Create everything
+    let sql = layout.as_ddl(None).unwrap();
+    assert!(sql.contains(BLOCK_IDX));
+    assert!(sql.contains(ATTR_IDX));
+
+    // Defer attribute indexes
+    let sql = layout.as_ddl(Some(index_list())).unwrap();
+    assert!(sql.contains(BLOCK_IDX));
+    assert!(!sql.contains(ATTR_IDX));
+    // This used to be duplicated
+    let count = sql.matches(BLOCK_IDX).count();
+    assert_eq!(1, count);
+
+    let table = layout.table(&SqlName::from("Data")).unwrap();
+    let sql = table.create_postponed_indexes(vec![], false);
+    assert_eq!(1, sql.len());
+    assert!(!sql[0].contains(BLOCK_IDX));
+    assert!(sql[0].contains(ATTR_IDX));
+
+    let dst_nsp = Namespace::new("sgd2".to_string()).unwrap();
+    let arr = index_list()
+        .indexes_for_table(&dst_nsp, &table.name.to_string(), &table, true, false)
+        .unwrap();
+    assert_eq!(1, arr.len());
+    assert!(!arr[0].1.contains(BLOCK_IDX));
+    assert!(arr[0].1.contains(ATTR_IDX));
+
+    let arr = index_list()
+        .indexes_for_table(&dst_nsp, &table.name.to_string(), &table, false, false)
+        .unwrap();
+    assert_eq!(0, arr.len());
+}
+
 const THING_GQL: &str = r#"
         type Thing @entity {
             id: ID!
@@ -1109,3 +1177,15 @@ on "sgd0815"."stats_3_day" using btree("volume");
 create index stats_3_day_dims
 on "sgd0815"."stats_3_day"(group_2, group_1, timestamp);
 "#;
+
+const BLOCK_GQL: &str = r#"
+type Block @entity(immutable: true) {
+    id: ID!
+    number: Int!
+}
+
+type Data @entity(immutable: true) {
+    id: ID!
+    block: Block!
+}
+"#;

store/postgres/src/relational/index.rs

@@ -123,7 +123,7 @@ impl Display for Expr {
             Expr::Column(s) => write!(f, "{s}")?,
             Expr::Prefix(s, _) => write!(f, "{s}")?,
             Expr::Vid => write!(f, "vid")?,
-            Expr::Block => write!(f, "block")?,
+            Expr::Block => write!(f, "{BLOCK_COLUMN}")?,
             Expr::BlockRange => write!(f, "block_range")?,
             Expr::BlockRangeLower => write!(f, "lower(block_range)")?,
             Expr::BlockRangeUpper => write!(f, "upper(block_range)")?,
@@ -488,12 +488,29 @@ impl CreateIndex {
                             && columns[1] == Expr::BlockRange
                     }
                     Method::Brin => false,
-                    Method::BTree | Method::Gin => {
+                    Method::Gin => {
+                        // 'using gin(<attr>)'
                         columns.len() == 1
                             && columns[0].is_attribute()
                             && cond.is_none()
                             && with.is_none()
                     }
+                    Method::BTree => {
+                        match columns.len() {
+                            1 => {
+                                // 'using btree(<attr>)'
+                                columns[0].is_attribute() && cond.is_none() && with.is_none()
+                            }
+                            2 => {
+                                // 'using btree(<attr>, block$)'
+                                columns[0].is_attribute()
+                                    && columns[1] == Expr::Block
+                                    && cond.is_none()
+                                    && with.is_none()
+                            }
+                            _ => false,
+                        }
+                    }
                     Method::Unknown(_) => false,
                 }
             }
@@ -537,6 +554,7 @@ impl CreateIndex {
                         None,
                     ),
                     dummy(false, BTree, &[Expr::BlockRangeUpper], Some(Cond::Closed)),
+                    dummy(false, BTree, &[Expr::Block], None),
                 ]
             };
         }
@@ -630,7 +648,7 @@ impl CreateIndex {
     }
 
     pub fn fields_exist_in_dest<'a>(&self, dest_table: &'a Table) -> bool {
-        fn column_exists<'a>(it: &mut impl Iterator<Item = &'a str>, column_name: &String) -> bool {
+        fn column_exists<'a>(it: &mut impl Iterator<Item = &'a str>, column_name: &str) -> bool {
             it.any(|c| *c == *column_name)
         }
 
@@ -667,9 +685,7 @@ impl CreateIndex {
                         }
                         Expr::Vid => (),
                         Expr::Block => {
-                            if !column_exists(cols, &"block".to_string()) {
-                                return false;
-                            }
+                            return dest_table.immutable;
                         }
                         Expr::Unknown(expression) => {
                             if some_column_contained(
@@ -776,7 +792,7 @@ impl IndexList {
                 // First we check if the fields do exist in the destination subgraph.
                 // In case of grafting that is not given.
                 if ci.fields_exist_in_dest(dest_table)
-                    // Then we check if the index is one of the default indexes not based on 
+                    // Then we check if the index is one of the default indexes not based on
                     // the attributes. Those will be created anyway and we should skip them.
                     && !ci.is_default_non_attr_index()
                     // Then ID based indexes in the immutable tables are also created initially