diff --git a/store/postgres/src/block_range.rs b/store/postgres/src/block_range.rs index d6877957dea..9b3e0ae39cc 100644 --- a/store/postgres/src/block_range.rs +++ b/store/postgres/src/block_range.rs @@ -96,25 +96,6 @@ impl ToSql, Pg> for BlockRange { } } -#[derive(Debug, Constructor)] -pub struct BlockRangeLowerBoundClause<'a> { - _table_prefix: &'a str, - block: BlockNumber, -} - -impl<'a> QueryFragment 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::(&self.block)?; - - Ok(()) - } -} - #[derive(Debug, Constructor)] pub struct BlockRangeUpperBoundClause<'a> { _table_prefix: &'a str, diff --git a/store/postgres/src/relational.rs b/store/postgres/src/relational.rs index 4059e736635..b04b13bf70e 100644 --- a/store/postgres/src/relational.rs +++ b/store/postgres/src/relational.rs @@ -699,9 +699,15 @@ impl Layout { let inserts_or_updates = FindChangesQuery::new(&tables[..], block) .load::(conn) .await?; - let deletions = FindPossibleDeletionsQuery::new(&tables[..], block) - .load::(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::(conn) + .await? + }; let mut processed_entities = HashSet::new(); let mut changes = Vec::new(); diff --git a/store/postgres/src/relational_queries.rs b/store/postgres/src/relational_queries.rs index d87b950adc1..a40a40b4f6c 100644 --- a/store/postgres/src/relational_queries.rs +++ b/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 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 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::(&self.block)?; } Ok(()) diff --git a/store/test-store/tests/postgres/relational.rs b/store/test-store/tests/postgres/relational.rs index c4946b408c5..fec176e9628 100644 --- a/store/test-store/tests/postgres/relational.rs +++ b/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, +) { + 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| {