add wrapping

Author: Zoran Cvetkov

chain/substreams/src/trigger.rs

@@ -8,6 +8,7 @@ use graph::{
         subgraph::{MappingError, ProofOfIndexingEvent, SharedProofOfIndexing},
         trigger_processor::HostedTrigger,
     },
+    data::store::EntityV,
     prelude::{
         anyhow, async_trait, BlockHash, BlockNumber, BlockState, CheapClone, RuntimeHostBuilder,
     },
@@ -225,7 +226,8 @@ where
                         logger,
                     );
 
-                    state.entity_cache.set(key, entity)?;
+                    // TODO: check if 0 is correct VID
+                    state.entity_cache.set(key, EntityV::new(entity, 0))?;
                 }
                 ParsedChanges::Delete(entity_key) => {
                     let entity_type = entity_key.entity_type.cheap_clone();

core/src/subgraph/runner.rs

@@ -15,6 +15,7 @@ use graph::components::{
     subgraph::{MappingError, PoICausalityRegion, ProofOfIndexing, SharedProofOfIndexing},
 };
 use graph::data::store::scalar::Bytes;
+use graph::data::store::EntityV;
 use graph::data::subgraph::{
     schema::{SubgraphError, SubgraphHealth},
     SubgraphFeature,
@@ -1584,7 +1585,8 @@ async fn update_proof_of_indexing(
             data.push((entity_cache.schema.poi_block_time(), block_time));
         }
         let poi = entity_cache.make_entity(data)?;
-        entity_cache.set(key, poi)
+        // VOI is autogenerated for POI table and our input is ignored
+        entity_cache.set(key, EntityV::new(poi, 0))
     }
 
     let _section_guard = stopwatch.start_section("update_proof_of_indexing");

graph/src/components/store/entity_cache.rs

@@ -7,7 +7,7 @@ use std::sync::Arc;
 use crate::cheap_clone::CheapClone;
 use crate::components::store::write::EntityModification;
 use crate::components::store::{self as s, Entity, EntityOperation};
-use crate::data::store::{EntityValidationError, Id, IdType, IntoEntityIterator};
+use crate::data::store::{EntityV, EntityValidationError, Id, IdType, IntoEntityIterator};
 use crate::prelude::ENV_VARS;
 use crate::schema::{EntityKey, InputSchema};
 use crate::util::intern::Error as InternError;
@@ -29,8 +29,8 @@ pub enum GetScope {
 #[derive(Debug, Clone)]
 enum EntityOp {
     Remove,
-    Update(Entity),
-    Overwrite(Entity),
+    Update(EntityV),
+    Overwrite(EntityV),
 }
 
 impl EntityOp {
@@ -41,7 +41,7 @@ impl EntityOp {
         use EntityOp::*;
         match (self, entity) {
             (Remove, _) => Ok(None),
-            (Overwrite(new), _) | (Update(new), None) => Ok(Some(new)),
+            (Overwrite(new), _) | (Update(new), None) => Ok(Some(new.e)),
             (Update(updates), Some(entity)) => {
                 let mut e = entity.borrow().clone();
                 e.merge_remove_null_fields(updates)?;
@@ -65,7 +65,7 @@ impl EntityOp {
         match self {
             // This is how `Overwrite` is constructed, by accumulating `Update` onto `Remove`.
             Remove => *self = Overwrite(update),
-            Update(current) | Overwrite(current) => current.merge(update),
+            Update(current) | Overwrite(current) => current.e.merge(update.e),
         }
     }
 }
@@ -278,9 +278,9 @@ impl EntityCache {
         ) -> Result<Option<Entity>, anyhow::Error> {
             match op {
                 EntityOp::Update(entity) | EntityOp::Overwrite(entity)
-                    if query.matches(key, entity) =>
+                    if query.matches(key, &entity.e) =>
                 {
-                    Ok(Some(entity.clone()))
+                    Ok(Some(entity.e.clone()))
                 }
                 EntityOp::Remove => Ok(None),
                 _ => Ok(None),
@@ -349,9 +349,9 @@ impl EntityCache {
     /// with existing data. The entity will be validated against the
     /// subgraph schema, and any errors will result in an `Err` being
     /// returned.
-    pub fn set(&mut self, key: EntityKey, entity: Entity) -> Result<(), anyhow::Error> {
+    pub fn set(&mut self, key: EntityKey, entity: EntityV) -> Result<(), anyhow::Error> {
         // check the validate for derived fields
-        let is_valid = entity.validate(&key).is_ok();
+        let is_valid = entity.e.validate(&key).is_ok();
 
         self.entity_op(key.clone(), EntityOp::Update(entity));
 
@@ -453,33 +453,33 @@ impl EntityCache {
         for (key, update) in self.updates {
             use EntityModification::*;
 
-            let is_poi = key.entity_type.is_poi();
+            // let is_poi = key.entity_type.is_poi();
             let current = self.current.remove(&key).and_then(|entity| entity);
             let modification = match (current, update) {
                 // Entity was created
                 (None, EntityOp::Update(mut updates))
                 | (None, EntityOp::Overwrite(mut updates)) => {
-                    updates.remove_null_fields();
+                    updates.e.remove_null_fields();
                     let data = Arc::new(updates);
-                    self.current.insert(key.clone(), Some(data.cheap_clone()));
-                    let vid = if is_poi { 0 } else { data.vid() };
+                    self.current
+                        .insert(key.clone(), Some(data.e.clone().into()));
                     Some(Insert {
                         key,
-                        data,
+                        data: data.e.clone().into(),
                         block,
                         end: None,
-                        vid,
+                        vid: data.vid,
                     })
                 }
                 // Entity may have been changed
                 (Some(current), EntityOp::Update(updates)) => {
                     let mut data = current.as_ref().clone();
+                    let vid = updates.vid;
                     data.merge_remove_null_fields(updates)
                         .map_err(|e| key.unknown_attribute(e))?;
                     let data = Arc::new(data);
                     self.current.insert(key.clone(), Some(data.cheap_clone()));
                     if current != data {
-                        let vid = if is_poi { 0 } else { data.vid() };
                         Some(Overwrite {
                             key,
                             data,
@@ -494,15 +494,15 @@ impl EntityCache {
                 // Entity was removed and then updated, so it will be overwritten
                 (Some(current), EntityOp::Overwrite(data)) => {
                     let data = Arc::new(data);
-                    self.current.insert(key.clone(), Some(data.clone()));
-                    if current != data {
-                        let vid = if is_poi { 0 } else { data.vid() };
+                    self.current
+                        .insert(key.clone(), Some(data.e.clone().into()));
+                    if current != data.e.clone().into() {
                         Some(Overwrite {
                             key,
-                            data,
+                            data: data.e.clone().into(),
                             block,
                             end: None,
-                            vid,
+                            vid: data.vid,
                         })
                     } else {
                         None

graph/src/components/store/mod.rs

@@ -30,7 +30,7 @@ use crate::cheap_clone::CheapClone;
 use crate::components::store::write::EntityModification;
 use crate::constraint_violation;
 use crate::data::store::scalar::Bytes;
-use crate::data::store::{Id, IdList, Value};
+use crate::data::store::{EntityV, Id, IdList, Value};
 use crate::data::value::Word;
 use crate::data_source::CausalityRegion;
 use crate::derive::CheapClone;
@@ -829,7 +829,7 @@ where
 pub enum EntityOperation {
     /// Locates the entity specified by `key` and sets its attributes according to the contents of
     /// `data`.  If no entity exists with this key, creates a new entity.
-    Set { key: EntityKey, data: Entity },
+    Set { key: EntityKey, data: EntityV },
 
     /// Removes an entity with the specified key, if one exists.
     Remove { key: EntityKey },

graph/src/data/store/mod.rs

@@ -913,6 +913,7 @@ impl Entity {
         Id::try_from(self.get("id").unwrap().clone()).expect("the id is set to a valid value")
     }
 
+    // TODO: only for tests!
     pub fn vid(&self) -> i64 {
         self.get("vid")
             .expect("the vid is set")
@@ -934,8 +935,8 @@ impl Entity {
     /// If a key exists in both entities, the value from `update` is chosen.
     /// If a key only exists on one entity, the value from that entity is chosen.
     /// If a key is set to `Value::Null` in `update`, the key/value pair is removed.
-    pub fn merge_remove_null_fields(&mut self, update: Entity) -> Result<(), InternError> {
-        for (key, value) in update.0.into_iter() {
+    pub fn merge_remove_null_fields(&mut self, update: EntityV) -> Result<(), InternError> {
+        for (key, value) in update.e.0.into_iter() {
             match value {
                 Value::Null => self.0.remove(&key),
                 _ => self.0.insert(&key, value)?,
@@ -1086,6 +1087,19 @@ impl std::fmt::Debug for Entity {
     }
 }
 
+/// An entity is represented as a map of attribute names to values.
+#[derive(Debug, Clone, CacheWeight, PartialEq, Eq, Serialize)]
+pub struct EntityV {
+    pub e: Entity,
+    pub vid: i64,
+}
+
+impl EntityV {
+    pub fn new(e: Entity, vid: i64) -> Self {
+        Self { e, vid }
+    }
+}
+
 /// An object that is returned from a query. It's a an `r::Value` which
 /// carries the attributes of the object (`__typename`, `id` etc.) and
 /// possibly a pointer to its parent if the query that constructed it is one

runtime/test/src/test.rs

@@ -478,13 +478,12 @@ async fn test_ipfs_block() {
 const USER_DATA: &str = "user_data";
 
 fn make_thing(id: &str, value: &str, vid: i64) -> (String, EntityModification) {
-    const DOCUMENT: &str =
-        " type Thing @entity { id: String!, value: String!, extra: String, vid: Int8 }";
+    const DOCUMENT: &str = " type Thing @entity { id: String!, value: String!, extra: String }";
     lazy_static! {
         static ref SCHEMA: InputSchema = InputSchema::raw(DOCUMENT, "doesntmatter");
         static ref THING_TYPE: EntityType = SCHEMA.entity_type("Thing").unwrap();
     }
-    let data = entity! { SCHEMA => id: id, value: value, extra: USER_DATA, vid:vid };
+    let data = entity! { SCHEMA => id: id, value: value, extra: USER_DATA};
     let key = THING_TYPE.parse_key(id).unwrap();
     (
         format!("{{ \"id\": \"{}\", \"value\": \"{}\"}}", id, value),

runtime/wasm/src/host_exports.rs

@@ -18,7 +18,7 @@ use graph::components::store::{EnsLookup, GetScope, LoadRelatedRequest};
 use graph::components::subgraph::{
     InstanceDSTemplate, PoICausalityRegion, ProofOfIndexingEvent, SharedProofOfIndexing,
 };
-use graph::data::store::{self};
+use graph::data::store::{self, EntityV};
 use graph::data_source::{CausalityRegion, DataSource, EntityTypeAccess};
 use graph::ensure;
 use graph::prelude::ethabi::param_type::Reader;
@@ -315,7 +315,7 @@ impl HostExports {
                 data.insert(store::ID.clone(), value);
             }
         }
-        data.insert(store::VID.clone(), Value::Int8(vid));
+        // data.insert(store::VID.clone(), Value::Int8(vid));
 
         self.check_invalid_fields(
             self.data_source.api_version.clone(),
@@ -352,7 +352,7 @@ impl HostExports {
 
         state.metrics.track_entity_write(&entity_type, &entity);
 
-        state.entity_cache.set(key, entity)?;
+        state.entity_cache.set(key, EntityV::new(entity, vid))?;
 
         Ok(())
     }

server/index-node/src/resolver.rs

@@ -768,7 +768,7 @@ fn entity_changes_to_graphql(entity_changes: Vec<EntityOperation>) -> r::Value {
                     .push(key.entity_id);
             }
             EntityOperation::Set { key, data } => {
-                updates.entry(key.entity_type).or_default().push(data);
+                updates.entry(key.entity_type).or_default().push(data.e);
             }
         }
     }

store/postgres/src/relational.rs

@@ -67,7 +67,7 @@ use crate::{
     },
 };
 use graph::components::store::{AttributeNames, DerivedEntityQuery};
-use graph::data::store::{Id, IdList, IdType, BYTES_SCALAR};
+use graph::data::store::{EntityV, Id, IdList, IdType, BYTES_SCALAR};
 use graph::data::subgraph::schema::POI_TABLE;
 use graph::prelude::{
     anyhow, info, BlockNumber, DeploymentHash, Entity, EntityChange, EntityOperation, Logger,
@@ -588,12 +588,12 @@ impl Layout {
 
         for entity_data in inserts_or_updates.into_iter() {
             let entity_type = entity_data.entity_type(&self.input_schema);
-            let data: Entity = entity_data.deserialize_with_layout(self, None)?;
-            let entity_id = data.id();
+            let data: EntityV = entity_data.deserialize_with_layout(self, None)?;
+            let entity_id = data.e.id();
             processed_entities.insert((entity_type.clone(), entity_id.clone()));
 
             changes.push(EntityOperation::Set {
-                key: entity_type.key_in(entity_id, CausalityRegion::from_entity(&data)),
+                key: entity_type.key_in(entity_id, CausalityRegion::from_entity(&data.e)),
                 data,
             });
         }

store/postgres/src/relational_queries.rs

@@ -14,7 +14,7 @@ use diesel::sql_types::{Array, BigInt, Binary, Bool, Int8, Integer, Jsonb, Text,
 use diesel::QuerySource as _;
 use graph::components::store::write::{EntityWrite, RowGroup, WriteChunk};
 use graph::components::store::{Child as StoreChild, DerivedEntityQuery};
-use graph::data::store::{Id, IdType, NULL};
+use graph::data::store::{EntityV, Id, IdType, NULL};
 use graph::data::store::{IdList, IdRef, QueryObject};
 use graph::data::subgraph::schema::POI_TABLE;
 use graph::data::value::{Object, Word};
@@ -174,6 +174,23 @@ impl FromEntityData for Entity {
     }
 }
 
+impl FromEntityData for EntityV {
+    const WITH_INTERNAL_KEYS: bool = false;
+
+    type Value = graph::prelude::Value;
+
+    fn from_data<I: Iterator<Item = Result<(Word, Self::Value), StoreError>>>(
+        schema: &InputSchema,
+        parent_id: Option<Id>,
+        iter: I,
+    ) -> Result<Self, StoreError> {
+        debug_assert_eq!(None, parent_id);
+        let e = schema.try_make_entity(iter).map_err(StoreError::from)?;
+        let vid = e.vid();
+        Ok(EntityV::new(e, vid))
+    }
+}
+
 impl FromEntityData for QueryObject {
     const WITH_INTERNAL_KEYS: bool = true;