store: Support immutable entities in entity changes

store/postgres/src/block_range.rs

@@ -96,25 +96,6 @@ impl ToSql<Range<Integer>, Pg> for BlockRange {
     }
 }
 
-#[derive(Debug, Constructor)]
-pub struct BlockRangeLowerBoundClause<'a> {
-    _table_prefix: &'a str,
-    block: BlockNumber,
-}
-
-impl<'a> QueryFragment<Pg> for BlockRangeLowerBoundClause<'a> {
-    fn walk_ast<'b>(&'b self, mut out: AstPass<'_, 'b, Pg>) -> QueryResult<()> {
-        out.unsafe_to_cache_prepared();
-
-        out.push_sql("lower(");
-        out.push_identifier(BLOCK_RANGE_COLUMN)?;
-        out.push_sql(") = ");
-        out.push_bind_param::<Integer, _>(&self.block)?;
-
-        Ok(())
-    }
-}
-
 #[derive(Debug, Constructor)]
 pub struct BlockRangeUpperBoundClause<'a> {
     _table_prefix: &'a str,

store/postgres/src/relational.rs

@@ -699,9 +699,15 @@ impl Layout {
         let inserts_or_updates = FindChangesQuery::new(&tables[..], block)
             .load::<EntityData>(conn)
             .await?;
-        let deletions = FindPossibleDeletionsQuery::new(&tables[..], block)
-            .load::<EntityDeletion>(conn)
-            .await?;
+        // Immutable entities have no block range and cannot be deleted.
+        tables.retain(|table| !table.immutable);
+        let deletions = if tables.is_empty() {
+            Vec::new()
+        } else {
+            FindPossibleDeletionsQuery::new(&tables[..], block)
+                .load::<EntityDeletion>(conn)
+                .await?
+        };
 
         let mut processed_entities = HashSet::new();
         let mut changes = Vec::new();

store/postgres/src/relational_queries.rs

@@ -50,8 +50,7 @@ use crate::relational::{
 use crate::{
     block_range::{
         BLOCK_COLUMN, BLOCK_RANGE_COLUMN, BLOCK_RANGE_CURRENT, BlockRangeColumn,
-        BlockRangeLowerBoundClause, BlockRangeUpperBoundClause, BlockRangeValue,
-        CAUSALITY_REGION_COLUMN,
+        BlockRangeUpperBoundClause, BlockRangeValue, CAUSALITY_REGION_COLUMN,
     },
     primary::Site,
 };
@@ -2005,17 +2004,17 @@ impl<'a, Conn> RunQueryDsl<Conn> for FindRangeQuery<'a> {}
 
 /// Builds a query over a given set of [`Table`]s in an attempt to find updated
 /// and/or newly inserted entities at a given block number; i.e. such that the
-/// block range's lower bound is equal to said block number.
+/// block range's lower bound, or the creation block for immutable entities,
+/// is equal to said block number.
 #[derive(Debug)]
 pub struct FindChangesQuery<'a> {
     pub(crate) tables: &'a [&'a Table],
-    br_clause: BlockRangeLowerBoundClause<'a>,
+    block: BlockNumber,
 }
 
 impl<'a> FindChangesQuery<'a> {
     pub fn new(tables: &'a [&'a Table], block: BlockNumber) -> Self {
-        let br_clause = BlockRangeLowerBoundClause::new("e.", block);
-        Self { tables, br_clause }
+        Self { tables, block }
     }
 }
 
@@ -2033,7 +2032,15 @@ impl<'a> QueryFragment<Pg> for FindChangesQuery<'a> {
             out.push_sql("  from ");
             out.push_sql(table.qualified_name.as_str());
             out.push_sql(" e\n where ");
-            self.br_clause.walk_ast(out.reborrow())?;
+            if table.immutable {
+                out.push_identifier(BLOCK_COLUMN)?;
+            } else {
+                out.push_sql("lower(");
+                out.push_identifier(BLOCK_RANGE_COLUMN)?;
+                out.push_sql(")");
+            }
+            out.push_sql(" = ");
+            out.push_bind_param::<Integer, _>(&self.block)?;
         }
 
         Ok(())

store/test-store/tests/postgres/relational.rs

@@ -1,5 +1,6 @@
 //! Test mapping of GraphQL schema to a relational schema
 use diesel_async::SimpleAsyncConnection;
+use graph::components::store::EntityOperation;
 use graph::components::store::write::{EntityModification, RowGroup};
 use graph::data::store::scalar;
 use graph::entity;
@@ -505,7 +506,11 @@ async fn insert_pets(conn: &mut AsyncPgConnection, layout: &Layout) {
 }
 
 async fn create_schema(conn: &mut AsyncPgConnection) -> Layout {
-    let schema = InputSchema::parse_latest(THINGS_GQL, THINGS_SUBGRAPH_ID.clone()).unwrap();
+    create_schema_from(conn, THINGS_GQL).await
+}
+
+async fn create_schema_from(conn: &mut AsyncPgConnection, gql: &str) -> Layout {
+    let schema = InputSchema::parse_latest(gql, THINGS_SUBGRAPH_ID.clone()).unwrap();
     let site = make_dummy_site(
         THINGS_SUBGRAPH_ID.clone(),
         NAMESPACE.clone(),
@@ -575,6 +580,141 @@ where
     .await;
 }
 
+async fn assert_changes_at(
+    conn: &mut AsyncPgConnection,
+    layout: &Layout,
+    block: BlockNumber,
+    expected: Vec<EntityOperation>,
+) {
+    let mut changes = layout.find_changes(conn, block).await.unwrap();
+    for change in &mut changes {
+        if let EntityOperation::Set { data, .. } = change {
+            data.remove_null_fields();
+        }
+    }
+    assert_eq!(changes.len(), expected.len(), "changes at block {block}");
+    for change in expected {
+        assert!(
+            changes.contains(&change),
+            "missing {change:?} in changes at block {block}: {changes:?}"
+        );
+    }
+}
+
+#[graph::test]
+async fn find_changes_with_immutable_entities() {
+    run_test_with_conn(async |conn| {
+        remove_schema(conn).await;
+        let layout = &create_schema_from(
+            conn,
+            "type Ferret @entity { id: ID!, name: String! }
+             type Mink @entity(immutable: true) { id: ID!, order: Int! }",
+        )
+        .await;
+        let ferret_type = layout.input_schema.entity_type("Ferret").unwrap();
+        let mink_type = layout.input_schema.entity_type("Mink").unwrap();
+        let old = entity! { layout.input_schema => id: "changed", name: "old", vid: 0i64 };
+        let removed = entity! { layout.input_schema => id: "removed", name: "gone", vid: 1i64 };
+        let updated = entity! { layout.input_schema => id: "changed", name: "new", vid: 2i64 };
+        let first = entity! { layout.input_schema => id: "first", order: 1, vid: 3i64 };
+        let later = entity! { layout.input_schema => id: "later", order: 2, vid: 4i64 };
+
+        insert_entity_at(
+            conn,
+            layout,
+            &ferret_type,
+            vec![old.clone(), removed.clone()],
+            0,
+        )
+        .await;
+        update_entity_at(conn, layout, &ferret_type, vec![updated.clone()], 1).await;
+        let removed_key = ferret_type.key(removed.id());
+        let group = row_group_delete(&ferret_type, 1, vec![removed_key.clone()]);
+        assert_eq!(
+            layout.delete(conn, &group, &MOCK_STOPWATCH).await.unwrap(),
+            1
+        );
+        insert_entity_at(conn, layout, &mink_type, vec![first.clone()], 1).await;
+        insert_entity_at(conn, layout, &mink_type, vec![later.clone()], 2).await;
+
+        assert_changes_at(
+            conn,
+            layout,
+            0,
+            vec![
+                EntityOperation::Set {
+                    key: ferret_type.key(old.id()),
+                    data: old,
+                },
+                EntityOperation::Set {
+                    key: removed_key.clone(),
+                    data: removed,
+                },
+            ],
+        )
+        .await;
+        assert_changes_at(
+            conn,
+            layout,
+            1,
+            vec![
+                EntityOperation::Set {
+                    key: ferret_type.key(updated.id()),
+                    data: updated,
+                },
+                EntityOperation::Set {
+                    key: mink_type.key(first.id()),
+                    data: first,
+                },
+                EntityOperation::Remove { key: removed_key },
+            ],
+        )
+        .await;
+        assert_changes_at(
+            conn,
+            layout,
+            2,
+            vec![EntityOperation::Set {
+                key: mink_type.key(later.id()),
+                data: later,
+            }],
+        )
+        .await;
+        assert_changes_at(conn, layout, 3, vec![]).await;
+    })
+    .await;
+}
+
+#[graph::test]
+async fn find_changes_with_only_immutable_entities() {
+    run_test_with_conn(async |conn| {
+        remove_schema(conn).await;
+        let layout = create_schema_from(
+            conn,
+            "type Event @entity(immutable: true) { id: ID!, value: Int! }",
+        )
+        .await;
+        let event_type = layout.input_schema.entity_type("Event").unwrap();
+        assert_changes_at(conn, &layout, 0, vec![]).await;
+
+        let event = entity! { layout.input_schema => id: "event", value: 42, vid: 0i64 };
+        insert_entity_at(conn, &layout, &event_type, vec![event.clone()], 1).await;
+        assert_changes_at(conn, &layout, 0, vec![]).await;
+        assert_changes_at(
+            conn,
+            &layout,
+            1,
+            vec![EntityOperation::Set {
+                key: event_type.key(event.id()),
+                data: event,
+            }],
+        )
+        .await;
+        assert_changes_at(conn, &layout, 2, vec![]).await;
+    })
+    .await;
+}
+
 #[graph::test]
 async fn find() {
     run_test(async |conn, layout| {