add vid

Author: Zoran Cvetkov

graph/src/components/store/entity_cache.rs

@@ -461,11 +461,13 @@ impl EntityCache {
                     updates.remove_null_fields();
                     let data = Arc::new(updates);
                     self.current.insert(key.clone(), Some(data.cheap_clone()));
+                    let vid = data.vid_opt().unwrap_or_default();
                     Some(Insert {
                         key,
                         data,
                         block,
                         end: None,
+                        vid,
                     })
                 }
                 // Entity may have been changed
@@ -476,11 +478,13 @@ impl EntityCache {
                     let data = Arc::new(data);
                     self.current.insert(key.clone(), Some(data.cheap_clone()));
                     if current != data {
+                        let vid = data.vid_opt().unwrap_or_default();
                         Some(Overwrite {
                             key,
                             data,
                             block,
                             end: None,
+                            vid,
                         })
                     } else {
                         None
@@ -491,11 +495,13 @@ impl EntityCache {
                     let data = Arc::new(data);
                     self.current.insert(key.clone(), Some(data.clone()));
                     if current != data {
+                        let vid = data.vid();
                         Some(Overwrite {
                             key,
                             data,
                             block,
                             end: None,
+                            vid,
                         })
                     } else {
                         None

graph/src/components/store/write.rs

@@ -45,13 +45,15 @@ pub enum EntityModification {
         data: Arc<Entity>,
         block: BlockNumber,
         end: Option<BlockNumber>,
+        vid: i64,
     },
     /// Update the entity by overwriting it
     Overwrite {
         key: EntityKey,
         data: Arc<Entity>,
         block: BlockNumber,
         end: Option<BlockNumber>,
+        vid: i64,
     },
     /// Remove the entity
     Remove { key: EntityKey, block: BlockNumber },
@@ -67,6 +69,7 @@ pub struct EntityWrite<'a> {
     // The end of the block range for which this write is valid. The value
     // of `end` itself is not included in the range
     pub end: Option<BlockNumber>,
+    pub vid: i64,
 }
 
 impl std::fmt::Display for EntityWrite<'_> {
@@ -89,24 +92,28 @@ impl<'a> TryFrom<&'a EntityModification> for EntityWrite<'a> {
                 data,
                 block,
                 end,
+                vid,
             } => Ok(EntityWrite {
                 id: &key.entity_id,
                 entity: data,
                 causality_region: key.causality_region,
                 block: *block,
                 end: *end,
+                vid: *vid,
             }),
             EntityModification::Overwrite {
                 key,
                 data,
                 block,
                 end,
+                vid,
             } => Ok(EntityWrite {
                 id: &key.entity_id,
                 entity: &data,
                 causality_region: key.causality_region,
                 block: *block,
                 end: *end,
+                vid: *vid,
             }),
 
             EntityModification::Remove { .. } => Err(()),
@@ -213,11 +220,13 @@ impl EntityModification {
                 data,
                 block,
                 end,
+                vid,
             } => Ok(Insert {
                 key,
                 data,
                 block,
                 end,
+                vid,
             }),
             Remove { key, .. } => {
                 return Err(constraint_violation!(
@@ -271,21 +280,23 @@ impl EntityModification {
 }
 
 impl EntityModification {
-    pub fn insert(key: EntityKey, data: Entity, block: BlockNumber) -> Self {
+    pub fn insert(key: EntityKey, data: Entity, block: BlockNumber, vid: i64) -> Self {
         EntityModification::Insert {
             key,
             data: Arc::new(data),
             block,
             end: None,
+            vid,
         }
     }
 
-    pub fn overwrite(key: EntityKey, data: Entity, block: BlockNumber) -> Self {
+    pub fn overwrite(key: EntityKey, data: Entity, block: BlockNumber, vid: i64) -> Self {
         EntityModification::Overwrite {
             key,
             data: Arc::new(data),
             block,
             end: None,
+            vid,
         }
     }
 
@@ -1017,31 +1028,36 @@ mod test {
 
             let value = value.clone();
             let key = THING_TYPE.parse_key("one").unwrap();
+            let vid = 0;
             match value {
                 Ins(block) => EntityModification::Insert {
                     key,
                     data: Arc::new(entity! { SCHEMA => id: "one", count: block }),
                     block,
                     end: None,
+                    vid,
                 },
                 Ovw(block) => EntityModification::Overwrite {
                     key,
                     data: Arc::new(entity! { SCHEMA => id: "one", count: block }),
                     block,
                     end: None,
+                    vid,
                 },
                 Rem(block) => EntityModification::Remove { key, block },
                 InsC(block, end) => EntityModification::Insert {
                     key,
                     data: Arc::new(entity! { SCHEMA => id: "one", count: block }),
                     block,
                     end: Some(end),
+                    vid,
                 },
                 OvwC(block, end) => EntityModification::Overwrite {
                     key,
                     data: Arc::new(entity! { SCHEMA => id: "one", count: block }),
                     block,
                     end: Some(end),
+                    vid,
                 },
             }
         }

graph/src/components/subgraph/instance.rs

@@ -77,6 +77,8 @@ pub struct BlockState {
     // data source that have been processed.
     pub processed_data_sources: Vec<StoredDynamicDataSource>,
 
+    pub vid_seq: i32,
+
     // Marks whether a handler is currently executing.
     in_handler: bool,
 
@@ -92,6 +94,7 @@ impl BlockState {
             persisted_data_sources: Vec::new(),
             handler_created_data_sources: Vec::new(),
             processed_data_sources: Vec::new(),
+            vid_seq: 0,
             in_handler: false,
             metrics: BlockStateMetrics::new(),
         }
@@ -109,6 +112,7 @@ impl BlockState {
             persisted_data_sources,
             handler_created_data_sources,
             processed_data_sources,
+            vid_seq: _,
             in_handler,
             metrics,
         } = self;
@@ -178,4 +182,9 @@ impl BlockState {
     pub fn persist_data_source(&mut self, ds: StoredDynamicDataSource) {
         self.persisted_data_sources.push(ds)
     }
+    pub fn next_vid(&mut self, block_number: BlockNumber) -> i64 {
+        let vid = ((block_number as i64) << 32) + self.vid_seq as i64;
+        self.vid_seq += 1;
+        vid
+    }
 }

graph/src/data/store/mod.rs

@@ -735,6 +735,9 @@ where
 lazy_static! {
     /// The name of the id attribute, `"id"`
     pub static ref ID: Word = Word::from("id");
+
+    /// The name of the vid attribute, `"vid"`
+    pub static ref VID: Word = Word::from("vid");
 }
 
 /// An entity is represented as a map of attribute names to values.
@@ -910,6 +913,17 @@ impl Entity {
         Id::try_from(self.get("id").unwrap().clone()).expect("the id is set to a valid value")
     }
 
+    pub fn vid(&self) -> i64 {
+        self.get("vid")
+            .expect("the vid is set")
+            .as_int8()
+            .expect("the vid is set to a valid value")
+    }
+
+    pub fn vid_opt(&self) -> Option<i64> {
+        self.get("vid").map(|vid| vid.as_int8()).unwrap_or_default()
+    }
+
     /// Merges an entity update `update` into this entity.
     ///
     /// If a key exists in both entities, the value from `update` is chosen.

graph/src/schema/input/mod.rs

@@ -35,6 +35,7 @@ pub(crate) const POI_OBJECT: &str = "Poi$";
 const POI_DIGEST: &str = "digest";
 /// The name of the PoI attribute for storing the block time
 const POI_BLOCK_TIME: &str = "blockTime";
+const VID: &str = "vid";
 
 pub mod kw {
     pub const ENTITY: &str = "entity";
@@ -1487,6 +1488,9 @@ impl InputSchema {
     }
 
     pub fn has_field_with_name(&self, entity_type: &EntityType, field: &str) -> bool {
+        if field == VID {
+            return true;
+        }
         let field = self.inner.pool.lookup(field);
 
         match field {
@@ -1597,6 +1601,8 @@ fn atom_pool(document: &s::Document) -> AtomPool {
     pool.intern(POI_DIGEST);
     pool.intern(POI_BLOCK_TIME);
 
+    pool.intern(VID);
+
     for definition in &document.definitions {
         match definition {
             s::Definition::TypeDefinition(typedef) => match typedef {

runtime/wasm/src/host_exports.rs

@@ -248,6 +248,7 @@ impl HostExports {
         gas: &GasCounter,
     ) -> Result<(), HostExportError> {
         let entity_type = state.entity_cache.schema.entity_type(&entity_type)?;
+        let vid = state.next_vid(block);
 
         Self::expect_object_type(&entity_type, "set")?;
 
@@ -314,6 +315,7 @@ impl HostExports {
                 data.insert(store::ID.clone(), value);
             }
         }
+        data.insert(store::VID.clone(), Value::Int8(vid));
 
         self.check_invalid_fields(
             self.data_source.api_version.clone(),

store/postgres/src/relational/ddl.rs

@@ -4,6 +4,7 @@ use std::{
 };
 
 use graph::{
+    data::subgraph::schema::POI_TABLE,
     prelude::{BLOCK_NUMBER_MAX, ENV_VARS},
     schema::InputSchema,
 };
@@ -116,19 +117,26 @@ impl Table {
             Ok(cols)
         }
 
+        let vid_type = if self.name.as_str() == POI_TABLE {
+            "bigserial"
+        } else {
+            "bigint"
+        };
+
         if self.immutable {
             writeln!(
                 out,
                 "
     create table {qname} (
-        {vid}                  bigserial primary key,
+        {vid}                  {vid_type} primary key,
         {block}                int not null,\n\
         {cols},
         unique({id})
     );",
                 qname = self.qualified_name,
                 cols = columns_ddl(self)?,
                 vid = VID_COLUMN,
+                vid_type = vid_type,
                 block = BLOCK_COLUMN,
                 id = self.primary_key().name
             )
@@ -137,13 +145,14 @@ impl Table {
                 out,
                 r#"
     create table {qname} (
-        {vid}                  bigserial primary key,
+        {vid}                  {vid_type} primary key,
         {block_range}          int4range not null,
         {cols}
     );"#,
                 qname = self.qualified_name,
                 cols = columns_ddl(self)?,
                 vid = VID_COLUMN,
+                vid_type = vid_type,
                 block_range = BLOCK_RANGE_COLUMN
             )?;
 

store/postgres/src/relational_queries.rs

@@ -15,6 +15,7 @@ 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::{IdList, IdRef, QueryObject};
+use graph::data::subgraph::schema::POI_TABLE;
 use graph::data::value::{Object, Word};
 use graph::data_source::CausalityRegion;
 use graph::prelude::{
@@ -2266,6 +2267,7 @@ struct InsertRow<'a> {
     values: Vec<InsertValue<'a>>,
     br_value: BlockRangeValue,
     causality_region: CausalityRegion,
+    vid: i64,
 }
 
 impl<'a> InsertRow<'a> {
@@ -2302,10 +2304,12 @@ impl<'a> InsertRow<'a> {
         }
         let br_value = BlockRangeValue::new(table, row.block, row.end);
         let causality_region = row.causality_region;
+        let vid = row.vid;
         Ok(Self {
             values,
             br_value,
             causality_region,
+            vid,
         })
     }
 }
@@ -2391,6 +2395,8 @@ impl<'a> QueryFragment<Pg> for InsertQuery<'a> {
         let out = &mut out;
         out.unsafe_to_cache_prepared();
 
+        let not_poi = self.table.name.as_str() != POI_TABLE;
+
         // Construct a query
         //   insert into schema.table(column, ...)
         //   values
@@ -2416,6 +2422,9 @@ impl<'a> QueryFragment<Pg> for InsertQuery<'a> {
             out.push_sql(CAUSALITY_REGION_COLUMN);
         };
 
+        if not_poi {
+            out.push_sql(", vid");
+        }
         out.push_sql(") values\n");
 
         for (i, row) in self.rows.iter().enumerate() {
@@ -2433,6 +2442,10 @@ impl<'a> QueryFragment<Pg> for InsertQuery<'a> {
                 out.push_sql(", ");
                 out.push_bind_param::<Integer, _>(&row.causality_region)?;
             };
+            if not_poi {
+                out.push_sql(", ");
+                out.push_bind_param::<BigInt, _>(&row.vid)?;
+            }
             out.push_sql(")");
         }