graph, store: Keep all pending changes in one large list

When we start combining batches, we will want to mutate some changes in
place to avoid the data copies that we would need if we kept them separated
by the kind of change that is needed.

Author: lutter

graph/src/components/store/write.rs

@@ -816,18 +816,21 @@ pub fn update_deployment_status(
         .map_err(StoreError::from)
 }
 
-/// Insert the errors and check if the subgraph needs to be set as unhealthy.
+/// Insert the errors and check if the subgraph needs to be set as
+/// unhealthy. The `latest_block` is only used to check whether the subgraph
+/// is healthy as of that block; errors are inserted according to the
+/// `block_ptr` they contain
 pub(crate) fn insert_subgraph_errors(
     conn: &PgConnection,
     id: &DeploymentHash,
     deterministic_errors: &[SubgraphError],
-    block: BlockNumber,
+    latest_block: BlockNumber,
 ) -> Result<(), StoreError> {
     for error in deterministic_errors {
         insert_subgraph_error(conn, error)?;
     }
 
-    check_health(conn, id, block)
+    check_health(conn, id, latest_block)
 }
 
 #[cfg(debug_assertions)]

store/postgres/src/deployment.rs

@@ -5,14 +5,15 @@ use diesel::prelude::*;
 use diesel::r2d2::{ConnectionManager, PooledConnection};
 use graph::anyhow::Context;
 use graph::blockchain::block_stream::FirehoseCursor;
-use graph::components::store::write::{EntityRef, EntityWrite, RowGroup, RowGroups};
+use graph::components::store::write::RowGroup;
 use graph::components::store::{
     Batch, DerivedEntityQuery, EntityKey, EntityType, PrunePhase, PruneReporter, PruneRequest,
     PruningStrategy, StoredDynamicDataSource, VersionStats,
 };
 use graph::components::versions::VERSIONS;
 use graph::data::query::Trace;
 use graph::data::subgraph::{status, SPEC_VERSION_0_0_6};
+use graph::data::value::Word;
 use graph::data_source::CausalityRegion;
 use graph::prelude::futures03::FutureExt;
 use graph::prelude::{
@@ -272,7 +273,8 @@ impl DeploymentStore {
         &self,
         conn: &PgConnection,
         layout: &Layout,
-        key: &EntityKey,
+        entity_type: &EntityType,
+        entity_id: &Word,
     ) -> Result<(), StoreError> {
         // Collect all types that share an interface implementation with this
         // entity type, and make sure there are no conflicting IDs.
@@ -291,100 +293,71 @@ impl DeploymentStore {
             .expect("API schema should be present")
             .clone();
         let types_for_interface = schema.types_for_interface();
-        let entity_type = key.entity_type.to_string();
+        let entity_type_str = entity_type.to_string();
         let types_with_shared_interface = Vec::from_iter(
             schema
-                .interfaces_for_type(&key.entity_type)
+                .interfaces_for_type(entity_type)
                 .into_iter()
                 .flatten()
                 .flat_map(|interface| &types_for_interface[&EntityType::from(interface)])
                 .map(EntityType::from)
-                .filter(|type_name| type_name != &key.entity_type),
+                .filter(|type_name| type_name != entity_type),
         );
 
         if !types_with_shared_interface.is_empty() {
             if let Some(conflicting_entity) =
-                layout.conflicting_entity(conn, &key.entity_id, types_with_shared_interface)?
+                layout.conflicting_entity(conn, entity_id, types_with_shared_interface)?
             {
                 return Err(StoreError::ConflictingId(
-                    entity_type,
-                    key.entity_id.to_string(),
+                    entity_type_str,
+                    entity_id.to_string(),
                     conflicting_entity,
                 ));
             }
         }
         Ok(())
     }
 
-    fn apply_entity_modifications(
+    fn apply_entity_modifications<'a>(
         &self,
         conn: &PgConnection,
         layout: &Layout,
-        inserts: &RowGroups<EntityWrite>,
-        overwrites: &RowGroups<EntityWrite>,
-        removes: &RowGroups<EntityRef>,
+        groups: impl Iterator<Item = &'a RowGroup>,
         stopwatch: &StopwatchMetrics,
     ) -> Result<i32, StoreError> {
         let mut count = 0;
 
-        // Apply modification groups.
-        // Inserts:
-        for group in &inserts.groups {
+        for group in groups {
+            if group.has_clamps() {
+                count -= self.remove_entities(group, conn, layout, stopwatch)? as i32;
+            }
             count += self.insert_entities(group, conn, layout, stopwatch)? as i32
         }
 
-        // Overwrites:
-        for group in &overwrites.groups {
-            // we do not update the count since the number of entities remains the same
-            self.overwrite_entities(group, conn, layout, stopwatch)?;
-        }
-
-        // Removals
-        for group in &removes.groups {
-            count -= self.remove_entities(group, conn, layout, stopwatch)? as i32;
-        }
         Ok(count)
     }
 
     fn insert_entities<'a>(
         &'a self,
-        group: &'a RowGroup<EntityWrite>,
+        group: &'a RowGroup,
         conn: &PgConnection,
         layout: &'a Layout,
         stopwatch: &StopwatchMetrics,
     ) -> Result<usize, StoreError> {
         let section = stopwatch.start_section("check_interface_entity_uniqueness");
-        for row in group.rows.iter() {
+        for row in group.writes() {
             // WARNING: This will potentially execute 2 queries for each entity key.
-            self.check_interface_entity_uniqueness(conn, layout, &row.key)?;
+            self.check_interface_entity_uniqueness(conn, layout, &group.entity_type, &row.id())?;
         }
         section.end();
 
         let _section = stopwatch.start_section("apply_entity_modifications_insert");
         layout.insert(conn, group, stopwatch)
     }
 
-    fn overwrite_entities<'a>(
-        &'a self,
-        group: &'a RowGroup<EntityWrite>,
-        conn: &PgConnection,
-        layout: &'a Layout,
-        stopwatch: &StopwatchMetrics,
-    ) -> Result<usize, StoreError> {
-        let section = stopwatch.start_section("check_interface_entity_uniqueness");
-        for row in group.rows.iter() {
-            // WARNING: This will potentially execute 2 queries for each entity key.
-            self.check_interface_entity_uniqueness(conn, layout, &row.key)?;
-        }
-        section.end();
-
-        let _section = stopwatch.start_section("apply_entity_modifications_update");
-        layout.update(conn, group, stopwatch)
-    }
-
     fn remove_entities(
         &self,
-        group: &RowGroup<EntityRef>,
+        group: &RowGroup,
         conn: &PgConnection,
         layout: &Layout,
         stopwatch: &StopwatchMetrics,
@@ -1137,9 +1110,7 @@ impl DeploymentStore {
                 let count = self.apply_entity_modifications(
                     &conn,
                     layout.as_ref(),
-                    &batch.inserts,
-                    &batch.overwrites,
-                    &batch.removes,
+                    batch.groups(),
                     stopwatch,
                 )?;
                 section.end();

store/postgres/src/deployment_store.rs

@@ -24,7 +24,7 @@ use diesel::types::{FromSql, ToSql};
 use diesel::{connection::SimpleConnection, Connection};
 use diesel::{debug_query, OptionalExtension, PgConnection, RunQueryDsl};
 use graph::cheap_clone::CheapClone;
-use graph::components::store::write::{EntityRef, EntityWrite, RowGroup};
+use graph::components::store::write::RowGroup;
 use graph::constraint_violation;
 use graph::data::graphql::TypeExt as _;
 use graph::data::query::Trace;
@@ -654,7 +654,7 @@ impl Layout {
     pub fn insert<'a>(
         &'a self,
         conn: &PgConnection,
-        group: &'a RowGroup<EntityWrite>,
+        group: &'a RowGroup,
         stopwatch: &StopwatchMetrics,
     ) -> Result<usize, StoreError> {
         let table = self.table_for_entity(&group.entity_type)?;
@@ -664,10 +664,13 @@ impl Layout {
         // We insert the entities in chunks to make sure each operation does
         // not exceed the maximum number of bindings allowed in queries
         let chunk_size = InsertQuery::chunk_size(table);
-        for chunk in group.rows.chunks(chunk_size) {
-            count += InsertQuery::new(table, chunk)?
-                .get_results(conn)
-                .map(|ids| ids.len())?
+        for chunk in group.write_chunks(chunk_size) {
+            // Empty chunks would lead to invalid SQL
+            if !chunk.is_empty() {
+                count += InsertQuery::new(table, &chunk)?
+                    .get_results(conn)
+                    .map(|ids| ids.len())?
+            }
         }
         Ok(count)
     }
@@ -800,15 +803,15 @@ impl Layout {
     pub fn update<'a>(
         &'a self,
         conn: &PgConnection,
-        group: &'a RowGroup<EntityWrite>,
+        group: &'a RowGroup,
         stopwatch: &StopwatchMetrics,
     ) -> Result<usize, StoreError> {
         let table = self.table_for_entity(&group.entity_type)?;
-        if table.immutable {
+        if table.immutable && group.has_clamps() {
             let ids = group
                 .rows
                 .iter()
-                .map(|row| row.key.entity_id.as_str())
+                .map(|row| row.id().as_str())
                 .collect::<Vec<_>>()
                 .join(", ");
             return Err(constraint_violation!(
@@ -819,9 +822,8 @@ impl Layout {
         }
 
         let section = stopwatch.start_section("update_modification_clamp_range_query");
-        for (block, rows) in group.runs() {
-            let entity_keys: Vec<&str> =
-                rows.iter().map(|row| row.key.entity_id.as_str()).collect();
+        for (block, rows) in group.clamps_by_block() {
+            let entity_keys: Vec<&str> = rows.iter().map(|row| row.id().as_str()).collect();
 
             ClampRangeQuery::new(table, &entity_keys, block)?.execute(conn)?;
         }
@@ -833,8 +835,8 @@ impl Layout {
         // We insert the entities in chunks to make sure each operation does
         // not exceed the maximum number of bindings allowed in queries
         let chunk_size = InsertQuery::chunk_size(table);
-        for chunk in group.rows.chunks(chunk_size) {
-            count += InsertQuery::new(table, chunk)?.execute(conn)?;
+        for chunk in group.write_chunks(chunk_size) {
+            count += InsertQuery::new(table, &chunk)?.execute(conn)?;
         }
 
         Ok(count)
@@ -843,24 +845,21 @@ impl Layout {
     pub fn delete(
         &self,
         conn: &PgConnection,
-        group: &RowGroup<EntityRef>,
+        group: &RowGroup,
         stopwatch: &StopwatchMetrics,
     ) -> Result<usize, StoreError> {
         let table = self.table_for_entity(&group.entity_type)?;
         if table.immutable {
             return Err(constraint_violation!(
                 "entities of type `{}` can not be deleted since they are immutable. Entity ids are [{}]",
-                table.object, group.rows.iter().map(|eref| &eref.key.entity_id).join(", ")
+                table.object, group.rows.iter().map(|eref| eref.id()).join(", ")
             ));
         }
 
         let _section = stopwatch.start_section("delete_modification_clamp_range_query");
         let mut count = 0;
-        for (block, rows) in group.runs() {
-            let ids: Vec<_> = rows
-                .iter()
-                .map(|eref| eref.key.entity_id.as_str())
-                .collect();
+        for (block, rows) in group.clamps_by_block() {
+            let ids: Vec<_> = rows.iter().map(|eref| eref.id().as_str()).collect();
             for chunk in ids.chunks(DELETE_OPERATION_CHUNK_SIZE) {
                 count += ClampRangeQuery::new(table, chunk, block)?.execute(conn)?
             }