all: Add an EntityVersion struct that contains an Entity and its vid

Author: lutter

chain/ethereum/src/network_indexer/network_indexer.rs

@@ -311,6 +311,7 @@ fn load_parent_block_from_store(context: &Context, block_ptr: BlockPtr) -> Block
         .and_then(move |block| {
             future::result(
                 block
+                    .data
                     .get("parent")
                     .ok_or_else(move || {
                         anyhow!("block {} has no parent", block_ptr_for_missing_parent,)

graph/src/components/store.rs

@@ -1034,7 +1034,7 @@ pub trait WritableStore: Send + Sync + 'static {
     async fn supports_proof_of_indexing(&self) -> Result<bool, StoreError>;
 
     /// Looks up an entity using the given store key at the latest block.
-    fn get(&self, key: &EntityKey) -> Result<Option<Entity>, StoreError>;
+    fn get(&self, key: &EntityKey) -> Result<Option<EntityVersion>, StoreError>;
 
     /// Transact the entity changes from a single block atomically into the store, and update the
     /// subgraph block pointer to `block_ptr_to`, and update the firehose cursor to `firehose_cursor`
@@ -1222,7 +1222,7 @@ impl WritableStore for MockStore {
         unimplemented!()
     }
 
-    fn get(&self, _: &EntityKey) -> Result<Option<Entity>, StoreError> {
+    fn get(&self, _: &EntityKey) -> Result<Option<EntityVersion>, StoreError> {
         unimplemented!()
     }
 
@@ -1769,7 +1769,7 @@ impl LfuCache<EntityKey, Option<Entity>> {
     ) -> Result<Option<Entity>, QueryExecutionError> {
         match self.get(key) {
             None => {
-                let mut entity = store.get(key)?;
+                let mut entity = store.get(key)?.map(|ev| ev.data);
                 if let Some(entity) = &mut entity {
                     // `__typename` is for queries not for mappings.
                     entity.remove("__typename");

graph/src/data/store/mod.rs

@@ -8,14 +8,14 @@ use itertools::Itertools;
 use serde::de;
 use serde::{Deserialize, Serialize};
 use stable_hash::prelude::*;
-use std::convert::TryFrom;
 use std::fmt;
 use std::iter::FromIterator;
 use std::str::FromStr;
 use std::{
     borrow::Cow,
     collections::{BTreeMap, HashMap},
 };
+use std::{convert::TryFrom, num::NonZeroU64};
 use strum::AsStaticRef as _;
 use strum_macros::AsStaticStr;
 
@@ -622,6 +622,33 @@ impl CacheWeight for Entity {
     }
 }
 
+pub type Vid = Option<NonZeroU64>;
+
+#[derive(Default, Debug)]
+pub struct EntityVersion {
+    pub data: Entity,
+    /// The `vid` of the entity if it exists in the store
+    pub vid: Vid,
+}
+
+impl EntityVersion {
+    pub fn new(data: Entity, vid: Vid) -> Self {
+        EntityVersion { data, vid }
+    }
+}
+
+impl From<EntityVersion> for Entity {
+    fn from(ev: EntityVersion) -> Self {
+        ev.data
+    }
+}
+
+impl From<Entity> for EntityVersion {
+    fn from(data: Entity) -> Self {
+        Self { data, vid: None }
+    }
+}
+
 /// A value that can (maybe) be converted to an `Entity`.
 pub trait TryIntoEntity {
     fn try_into_entity(self) -> Result<Entity, Error>;

store/postgres/src/deployment_store.rs

@@ -4,6 +4,7 @@ use diesel::pg::PgConnection;
 use diesel::prelude::*;
 use diesel::r2d2::{ConnectionManager, PooledConnection};
 use graph::components::store::{EntityType, StoredDynamicDataSource};
+use graph::data::store::EntityVersion;
 use graph::data::subgraph::status;
 use graph::prelude::{
     tokio, CancelHandle, CancelToken, CancelableError, PoolWaitStats, SubgraphDeploymentEntity,
@@ -801,7 +802,7 @@ impl DeploymentStore {
         &self,
         site: Arc<Site>,
         key: &EntityKey,
-    ) -> Result<Option<Entity>, StoreError> {
+    ) -> Result<Option<EntityVersion>, StoreError> {
         let conn = self.get_conn()?;
         let layout = self.layout(&conn, site)?;
 

store/postgres/src/relational.rs

@@ -32,7 +32,7 @@ use crate::{
 use graph::components::store::EntityType;
 use graph::data::graphql::ext::{DirectiveFinder, DocumentExt, ObjectTypeExt};
 use graph::data::schema::{FulltextConfig, FulltextDefinition, Schema, SCHEMA_TYPE_NAME};
-use graph::data::store::BYTES_SCALAR;
+use graph::data::store::{EntityVersion, BYTES_SCALAR};
 use graph::data::subgraph::schema::{POI_OBJECT, POI_TABLE};
 use graph::prelude::{
     anyhow, info, BlockNumber, DeploymentHash, Entity, EntityChange, EntityCollection,
@@ -522,7 +522,7 @@ impl Layout {
         entity: &EntityType,
         id: &str,
         block: BlockNumber,
-    ) -> Result<Option<Entity>, StoreError> {
+    ) -> Result<Option<EntityVersion>, StoreError> {
         let table = self.table_for_entity(entity)?;
         FindQuery::new(table.as_ref(), id, block)
             .get_result::<EntityData>(conn)

store/postgres/src/relational_queries.rs

@@ -11,6 +11,7 @@ use diesel::query_dsl::{LoadQuery, RunQueryDsl};
 use diesel::result::{Error as DieselError, QueryResult};
 use diesel::sql_types::{Array, BigInt, Binary, Bool, Integer, Jsonb, Range, Text};
 use diesel::Connection;
+use graph::data::store::{EntityVersion, Vid};
 use lazy_static::lazy_static;
 
 use graph::prelude::{
@@ -29,6 +30,7 @@ use std::convert::TryFrom;
 use std::env;
 use std::fmt::{self, Display};
 use std::iter::FromIterator;
+use std::num::NonZeroU64;
 use std::str::FromStr;
 
 use crate::relational::{
@@ -273,6 +275,8 @@ pub trait FromEntityData: Default {
     type Value: FromColumnValue;
 
     fn insert_entity_data(&mut self, key: String, v: Self::Value);
+
+    fn set_vid(&mut self, vid: Vid);
 }
 
 impl FromEntityData for Entity {
@@ -281,6 +285,10 @@ impl FromEntityData for Entity {
     fn insert_entity_data(&mut self, key: String, v: Self::Value) {
         self.insert(key, v);
     }
+
+    fn set_vid(&mut self, _vid: Vid) {
+        /* ignore */
+    }
 }
 
 impl FromEntityData for BTreeMap<String, r::Value> {
@@ -289,6 +297,22 @@ impl FromEntityData for BTreeMap<String, r::Value> {
     fn insert_entity_data(&mut self, key: String, v: Self::Value) {
         self.insert(key, v);
     }
+
+    fn set_vid(&mut self, _vid: Vid) {
+        /* ignore */
+    }
+}
+
+impl FromEntityData for EntityVersion {
+    type Value = graph::prelude::Value;
+
+    fn insert_entity_data(&mut self, key: String, v: Self::Value) {
+        self.data.insert(key, v);
+    }
+
+    fn set_vid(&mut self, vid: Vid) {
+        self.vid = vid
+    }
 }
 
 pub trait FromColumnValue: Sized {
@@ -461,6 +485,8 @@ impl FromColumnValue for graph::prelude::Value {
 pub struct EntityData {
     #[sql_type = "Text"]
     entity: String,
+    #[sql_type = "BigInt"]
+    vid: i64,
     #[sql_type = "Jsonb"]
     data: serde_json::Value,
 }
@@ -486,6 +512,7 @@ impl EntityData {
                     "__typename".to_owned(),
                     T::Value::from_string(entity_type.into_string()),
                 );
+                out.set_vid(NonZeroU64::new(self.vid as u64));
                 for (key, json) in map {
                     // Simply ignore keys that do not have an underlying table
                     // column; those will be things like the block_range that
@@ -1169,7 +1196,7 @@ impl<'a> QueryFragment<Pg> for FindQuery<'a> {
         //      from schema.table e where id = $1
         out.push_sql("select ");
         out.push_bind_param::<Text, _>(&self.table.object.as_str())?;
-        out.push_sql(" as entity, to_jsonb(e.*) as data\n");
+        out.push_sql(" as entity, e.vid, to_jsonb(e.*) as data\n");
         out.push_sql("  from ");
         out.push_sql(self.table.qualified_name.as_str());
         out.push_sql(" e\n where ");
@@ -1221,7 +1248,7 @@ impl<'a> QueryFragment<Pg> for FindManyQuery<'a> {
             }
             out.push_sql("select ");
             out.push_bind_param::<Text, _>(&table.object.as_str())?;
-            out.push_sql(" as entity, to_jsonb(e.*) as data\n");
+            out.push_sql(" as entity, e.vid, to_jsonb(e.*) as data\n");
             out.push_sql("  from ");
             out.push_sql(table.qualified_name.as_str());
             out.push_sql(" e\n where ");
@@ -2296,7 +2323,7 @@ impl<'a> FilterQuery<'a> {
     fn select_entity_and_data(table: &Table, out: &mut AstPass<Pg>) {
         out.push_sql("select '");
         out.push_sql(table.object.as_str());
-        out.push_sql("' as entity, to_jsonb(c.*) as data");
+        out.push_sql("' as entity, c.vid, to_jsonb(c.*) as data");
     }
 
     /// Only one table/filter pair, and no window
@@ -2412,7 +2439,7 @@ impl<'a> FilterQuery<'a> {
             if i > 0 {
                 out.push_sql("\nunion all\n");
             }
-            out.push_sql("select m.entity, ");
+            out.push_sql("select m.entity, m.vid, ");
             jsonb_build_object(column_names, "c", &table, &mut out)?;
             out.push_sql(" as data, c.id");
             self.sort_key.select(&mut out)?;

store/postgres/src/subgraph_store.rs

@@ -17,7 +17,7 @@ use graph::{
     constraint_violation,
     data::query::QueryTarget,
     data::subgraph::schema::SubgraphError,
-    data::subgraph::status,
+    data::{store::EntityVersion, subgraph::status},
     prelude::StoreEvent,
     prelude::SubgraphDeploymentEntity,
     prelude::{
@@ -1228,7 +1228,7 @@ impl WritableStoreTrait for WritableStore {
         .await
     }
 
-    fn get(&self, key: &EntityKey) -> Result<Option<Entity>, StoreError> {
+    fn get(&self, key: &EntityKey) -> Result<Option<EntityVersion>, StoreError> {
         self.retry("get", || self.writable.get(self.site.cheap_clone(), key))
     }
 

store/postgres/tests/relational.rs

@@ -437,7 +437,7 @@ macro_rules! assert_entity_eq {
         let mut pass = true;
 
         for (key, left_value) in left.clone().sorted() {
-            match right.get(&key) {
+            match right.data.get(&key) {
                 None => {
                     pass = false;
                     println!("key '{}' missing from right", key);
@@ -453,7 +453,7 @@ macro_rules! assert_entity_eq {
                 }
             }
         }
-        for (key, _) in right.clone().sorted() {
+        for (key, _) in right.data.clone().sorted() {
             if left.get(&key).is_none() {
                 pass = false;
                 println!("key '{}' missing from left", key);
@@ -619,6 +619,7 @@ fn update_many() {
                     .find(conn, &*SCALAR, id, BLOCK_NUMBER_MAX)
                     .expect(&format!("Failed to read Scalar[{}]", id))
                     .unwrap()
+                    .data
             })
             .collect();
         let new_one = &updated[0];

store/postgres/tests/relational_bytes.rs

@@ -145,7 +145,7 @@ macro_rules! assert_entity_eq {
         let mut pass = true;
 
         for (key, left_value) in left.clone().sorted() {
-            match right.get(&key) {
+            match right.data.get(&key) {
                 None => {
                     pass = false;
                     println!("key '{}' missing from right", key);
@@ -161,7 +161,7 @@ macro_rules! assert_entity_eq {
                 }
             }
         }
-        for (key, _) in right.clone().sorted() {
+        for (key, _) in right.data.clone().sorted() {
             if left.get(&key).is_none() {
                 pass = false;
                 println!("key '{}' missing from left", key);