deduce the types of operation create, modify or delete

Author: Zoran Cvetkov

store/postgres/src/relational.rs

@@ -53,8 +53,8 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
 use crate::relational_queries::{
-    ConflictingEntitiesData, ConflictingEntitiesQuery, FindChangesQuery, FindDerivedQuery,
-    FindPossibleDeletionsQuery, ReturnedEntityData,
+    ConflictingEntitiesData, ConflictingEntitiesQuery, EntityDataExt, FindChangesQuery,
+    FindDerivedQuery, FindPossibleDeletionsQuery, ReturnedEntityData,
 };
 use crate::{
     primary::{Namespace, Site},
@@ -529,28 +529,85 @@ impl Layout {
             et_map.insert(et.to_string(), Arc::new(et));
         }
         let mut entities: BTreeMap<BlockNumber, Vec<EntityWithType>> = BTreeMap::new();
-        if let Some(vec) = FindRangeQuery::new(&tables, false, block_range)
-            .get_results::<EntityData>(conn)
+        let lower_vec = FindRangeQuery::new(&tables, false, block_range.clone())
+            .get_results::<EntityDataExt>(conn)
             .optional()?
-        {
-            // TODO: issue query with upper range and find modifictions and deletions
-            for e in vec {
-                let block = e.clone().deserialize_block_number::<Entity>()?;
-                let entity_type = e.entity_type(&self.input_schema);
-                let entity = e.deserialize_with_layout::<Entity>(self, None)?;
-                let entity_op = EntitySubgraphOperation::Create;
-                let ewt = EntityWithType {
-                    entity_op,
-                    entity_type,
-                    entity,
-                };
-                match entities.get_mut(&block) {
-                    Some(vec) => vec.push(ewt),
-                    None => {
-                        let _ = entities.insert(block, vec![ewt]);
+            .unwrap_or_default();
+        let upper_vec = FindRangeQuery::new(&tables, true, block_range)
+            .get_results::<EntityDataExt>(conn)
+            .optional()?
+            .unwrap_or_default();
+        let mut lower_iter = lower_vec.iter().fuse().peekable();
+        let mut upper_iter = upper_vec.iter().fuse().peekable();
+        let mut lower_now = lower_iter.next();
+        let mut upper_now = upper_iter.next();
+        let mut lower = lower_now.unwrap_or(&EntityDataExt::default()).clone();
+        let mut upper = upper_now.unwrap_or(&EntityDataExt::default()).clone();
+        let transform = |ede: EntityDataExt,
+                         entity_op: EntitySubgraphOperation|
+         -> Result<(EntityWithType, BlockNumber), StoreError> {
+            let e = EntityData::new(ede.entity, ede.data);
+            let block = ede.block_number;
+            let entity_type = e.entity_type(&self.input_schema);
+            let entity = e.deserialize_with_layout::<Entity>(self, None)?;
+            let ewt = EntityWithType {
+                entity_op,
+                entity_type,
+                entity,
+            };
+            Ok((ewt, block))
+        };
+        while lower_now.is_some() || upper_now.is_some() {
+            let (ewt, block) = if lower_now.is_some() {
+                if upper_now.is_some() {
+                    if lower > upper {
+                        // we have upper bound at this block, but no lower bounds at the same block so it's deletion
+                        let (ewt, block) = transform(upper, EntitySubgraphOperation::Delete)?;
+                        // advance upper
+                        upper_now = upper_iter.next();
+                        upper = upper_now.unwrap_or(&EntityDataExt::default()).clone();
+                        (ewt, block)
+                    } else if lower < upper {
+                        // we have lower bound at this block but no upper bound at the same block so its creation
+                        let (ewt, block) = transform(lower, EntitySubgraphOperation::Create)?;
+                        // advance lower
+                        lower_now = lower_iter.next();
+                        lower = lower_now.unwrap_or(&EntityDataExt::default()).clone();
+                        (ewt, block)
+                    } else {
+                        assert!(upper == lower);
+                        let (ewt, block) = transform(lower, EntitySubgraphOperation::Modify)?;
+                        // advance both
+                        lower_now = lower_iter.next();
+                        lower = lower_now.unwrap_or(&EntityDataExt::default()).clone();
+                        upper_now = upper_iter.next();
+                        upper = upper_now.unwrap_or(&EntityDataExt::default()).clone();
+                        (ewt, block)
                     }
-                };
-            }
+                } else {
+                    // we have lower bound at this block but no upper bound at the same block so its creation
+                    let (ewt, block) = transform(lower, EntitySubgraphOperation::Create)?;
+                    // advance lower
+                    lower_now = lower_iter.next();
+                    lower = lower_now.unwrap_or(&EntityDataExt::default()).clone();
+                    (ewt, block)
+                }
+            } else {
+                // we have upper bound at this block, but no lower bounds at all so it's deletion
+                assert!(upper_now.is_some());
+                let (ewt, block) = transform(upper, EntitySubgraphOperation::Delete)?;
+                // advance upper
+                upper_now = upper_iter.next();
+                upper = upper_now.unwrap_or(&EntityDataExt::default()).clone();
+                (ewt, block)
+            };
+
+            match entities.get_mut(&block) {
+                Some(vec) => vec.push(ewt),
+                None => {
+                    let _ = entities.insert(block, vec![ewt]);
+                }
+            };
         }
         Ok(entities)
     }

store/postgres/src/relational_queries.rs

@@ -17,7 +17,6 @@ use graph::data::store::{Id, IdType, NULL};
 use graph::data::store::{IdList, IdRef, QueryObject};
 use graph::data::value::{Object, Word};
 use graph::data_source::CausalityRegion;
-use graph::prelude::regex::Regex;
 use graph::prelude::{
     anyhow, r, serde_json, BlockNumber, ChildMultiplicity, Entity, EntityCollection, EntityFilter,
     EntityLink, EntityOrder, EntityOrderByChild, EntityOrderByChildInfo, EntityRange, EntityWindow,
@@ -27,6 +26,7 @@ use graph::schema::{EntityKey, EntityType, FulltextAlgorithm, FulltextConfig, In
 use graph::{components::store::AttributeNames, data::store::scalar};
 use inflector::Inflector;
 use itertools::Itertools;
+use std::cmp::Ordering;
 use std::collections::{BTreeMap, BTreeSet, HashSet};
 use std::convert::TryFrom;
 use std::fmt::{self, Display};
@@ -478,40 +478,12 @@ pub struct EntityData {
 }
 
 impl EntityData {
-    pub fn entity_type(&self, schema: &InputSchema) -> EntityType {
-        schema.entity_type(&self.entity).unwrap()
+    pub fn new(entity: String, data: serde_json::Value) -> EntityData {
+        EntityData { entity, data }
     }
 
-    pub fn deserialize_block_number<T: FromEntityData>(self) -> Result<BlockNumber, StoreError> {
-        use serde_json::Value as j;
-        match self.data {
-            j::Object(map) => {
-                let mut entries = map.into_iter().filter_map(move |(key, json)| {
-                    if key == "block_range" {
-                        let r = json.as_str().unwrap();
-                        let rx = Regex::new("\\[(?P<start>[0-9]+),([0-9]+)?\\)").unwrap();
-                        let cap = rx.captures(r).unwrap();
-                        let start = cap
-                            .name("start")
-                            .map(|mtch| mtch.as_str().to_string())
-                            .unwrap();
-                        let n = start.parse::<i32>().unwrap();
-                        Some(n)
-                    } else if key == "block$" {
-                        let block = json.as_i64().unwrap() as i32;
-                        Some(block)
-                    } else {
-                        None
-                    }
-                });
-                let en = entries.next().unwrap();
-                assert!(entries.next().is_none()); // there should be just one block_range field
-                Ok(en)
-            }
-            _ => unreachable!(
-                "we use `to_json` in our queries, and will therefore always get an object back"
-            ),
-        }
+    pub fn entity_type(&self, schema: &InputSchema) -> EntityType {
+        schema.entity_type(&self.entity).unwrap()
     }
 
     /// Map the `EntityData` using the schema information in `Layout`
@@ -586,6 +558,46 @@ impl EntityData {
     }
 }
 
+#[derive(QueryableByName, Clone, Debug, Default, Eq)]
+pub struct EntityDataExt {
+    #[diesel(sql_type = Text)]
+    pub entity: String,
+    #[diesel(sql_type = Jsonb)]
+    pub data: serde_json::Value,
+    #[diesel(sql_type = Integer)]
+    pub block_number: i32,
+    #[diesel(sql_type = Text)]
+    pub id: String,
+}
+
+impl Ord for EntityDataExt {
+    fn cmp(&self, other: &Self) -> Ordering {
+        let ord = self.block_number.cmp(&other.block_number);
+        if ord != Ordering::Equal {
+            ord
+        } else {
+            let ord = self.entity.cmp(&other.entity);
+            if ord != Ordering::Equal {
+                ord
+            } else {
+                self.id.cmp(&other.id)
+            }
+        }
+    }
+}
+
+impl PartialOrd for EntityDataExt {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl PartialEq for EntityDataExt {
+    fn eq(&self, other: &Self) -> bool {
+        self.cmp(other) == Ordering::Equal
+    }
+}
+
 /// The equivalent of `graph::data::store::Value` but in a form that does
 /// not require further transformation during `walk_ast`. This form takes
 /// the idiosyncrasies of how we serialize values into account (e.g., that