query for both lower and upper range

Author: Zoran Cvetkov

graph/src/blockchain/block_stream.rs

@@ -426,7 +426,16 @@ async fn scan_subgraph_triggers<C: Blockchain>(
     }
 }
 
+#[derive(Debug)]
+pub enum EntitySubgraphOperation {
+    Create,
+    Modify,
+    Delete,
+}
+
+#[derive(Debug)]
 pub struct EntityWithType {
+    pub entity_op: EntitySubgraphOperation,
     pub entity_type: EntityType,
     pub entity: Entity,
 }

store/postgres/src/block_range.rs

@@ -135,27 +135,31 @@ impl<'a> QueryFragment<Pg> for BlockRangeUpperBoundClause<'a> {
 /// Helper for generating SQL fragments for selecting entities in a specific block range
 #[derive(Debug, Clone, Copy)]
 pub enum EntityBlockRange {
-    Mutable(BlockRange), // TODO: check if this is a proper type here (maybe Range<BlockNumber>?)
+    Mutable((BlockRange, bool)), // TODO: check if this is a proper type here (maybe Range<BlockNumber>?)
     Immutable(BlockRange),
 }
 
 impl EntityBlockRange {
-    pub fn new(immutable: bool, block_range: std::ops::Range<BlockNumber>) -> Self {
+    pub fn new(
+        immutable: bool,
+        block_range: std::ops::Range<BlockNumber>,
+        is_uppper_range: bool,
+    ) -> Self {
         let start: Bound<BlockNumber> = Bound::Included(block_range.start);
         let end: Bound<BlockNumber> = Bound::Excluded(block_range.end);
         let block_range: BlockRange = BlockRange(start, end);
         if immutable {
             Self::Immutable(block_range)
         } else {
-            Self::Mutable(block_range)
+            Self::Mutable((block_range, is_uppper_range))
         }
     }
 
     /// Output SQL that matches only rows whose block range contains `block`.
     pub fn contains<'b>(&'b self, out: &mut AstPass<'_, 'b, Pg>) -> QueryResult<()> {
         out.unsafe_to_cache_prepared();
         let block_range = match self {
-            EntityBlockRange::Mutable(br) => br,
+            EntityBlockRange::Mutable((br, _)) => br,
             EntityBlockRange::Immutable(br) => br,
         };
         let BlockRange(start, finish) = block_range;
@@ -186,7 +190,13 @@ impl EntityBlockRange {
 
     pub fn compare_column(&self, out: &mut AstPass<Pg>) {
         match self {
-            EntityBlockRange::Mutable(_) => out.push_sql(" lower(block_range) "),
+            EntityBlockRange::Mutable((_, is_upper_range)) => {
+                if *is_upper_range {
+                    out.push_sql(" upper(block_range) ")
+                } else {
+                    out.push_sql(" lower(block_range) ")
+                }
+            }
             EntityBlockRange::Immutable(_) => out.push_sql(" block$ "),
         }
     }

store/postgres/src/relational.rs

@@ -24,7 +24,7 @@ use diesel::serialize::{Output, ToSql};
 use diesel::sql_types::Text;
 use diesel::{connection::SimpleConnection, Connection};
 use diesel::{debug_query, sql_query, OptionalExtension, PgConnection, QueryResult, RunQueryDsl};
-use graph::blockchain::block_stream::EntityWithType;
+use graph::blockchain::block_stream::{EntitySubgraphOperation, EntityWithType};
 use graph::blockchain::BlockTime;
 use graph::cheap_clone::CheapClone;
 use graph::components::store::write::{RowGroup, WriteChunk};
@@ -529,15 +529,18 @@ 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, block_range)
+        if let Some(vec) = FindRangeQuery::new(&tables, false, block_range)
             .get_results::<EntityData>(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,
                 };

store/postgres/src/relational_queries.rs

@@ -2009,16 +2009,22 @@ impl<'a, Conn> RunQueryDsl<Conn> for FindQuery<'a> {}
 #[derive(Debug, Clone)]
 pub struct FindRangeQuery<'a> {
     tables: &'a Vec<&'a Table>,
+    is_upper_range: bool,
     imm_range: EntityBlockRange,
     mut_range: EntityBlockRange,
 }
 
 impl<'a> FindRangeQuery<'a> {
-    pub fn new(tables: &'a Vec<&Table>, block_range: Range<BlockNumber>) -> Self {
-        let imm_range = EntityBlockRange::new(true, block_range.clone());
-        let mut_range = EntityBlockRange::new(false, block_range);
+    pub fn new(
+        tables: &'a Vec<&Table>,
+        is_upper_range: bool,
+        block_range: Range<BlockNumber>,
+    ) -> Self {
+        let imm_range = EntityBlockRange::new(true, block_range.clone(), false);
+        let mut_range = EntityBlockRange::new(false, block_range, is_upper_range);
         Self {
             tables,
+            is_upper_range,
             imm_range,
             mut_range,
         }
@@ -2028,42 +2034,46 @@ impl<'a> FindRangeQuery<'a> {
 impl<'a> QueryFragment<Pg> for FindRangeQuery<'a> {
     fn walk_ast<'b>(&'b self, mut out: AstPass<'_, 'b, Pg>) -> QueryResult<()> {
         out.unsafe_to_cache_prepared();
+        let mut first = true;
 
-        let mut iter = self.tables.iter().peekable();
-        while let Some(table) = iter.next() {
-            // Generate
-            //    select '..' as entity, to_jsonb(e.*) as data, block$ as block_number
-            //      from schema.table e where id = $1
-            out.push_sql("select ");
-            out.push_bind_param::<Text, _>(table.object.as_str())?;
-            out.push_sql(" as entity, to_jsonb(e.*) as data,");
-            if table.immutable {
-                self.imm_range.compare_column(&mut out)
-            } else {
-                self.mut_range.compare_column(&mut out)
-            }
-            out.push_sql("as block_number\n");
-            out.push_sql("  from ");
-            out.push_sql(table.qualified_name.as_str());
-            out.push_sql(" e\n  where");
-            // TODO: do we need to care about it?
-            // if self.table.has_causality_region {
-            //     out.push_sql("causality_region = ");
-            //     out.push_bind_param::<Integer, _>(&self.key.causality_region)?;
-            //     out.push_sql(" and ");
-            // }
-            if table.immutable {
-                self.imm_range.contains(&mut out)?;
-            } else {
-                self.mut_range.contains(&mut out)?;
-            }
-            // more elements left?
-            if iter.peek().is_some() {
-                out.push_sql("\nunion all\n"); // with the next
-            } else {
-                out.push_sql("\norder by block_number"); // on the last
+        for table in self.tables.iter() {
+            // the immutable entities don't have upper range and also can't be modified or deleted
+            if !(self.is_upper_range && table.immutable) {
+                if first {
+                    first = false;
+                } else {
+                    out.push_sql("\nunion all\n");
+                }
+
+                // Generate
+                //    select '..' as entity, to_jsonb(e.*) as data, block$ as block_number
+                //      from schema.table e where id = $1
+                out.push_sql("select ");
+                out.push_bind_param::<Text, _>(table.object.as_str())?;
+                out.push_sql(" as entity, to_jsonb(e.*) as data,");
+                if table.immutable {
+                    self.imm_range.compare_column(&mut out)
+                } else {
+                    self.mut_range.compare_column(&mut out)
+                }
+                out.push_sql("as block_number, id\n");
+                out.push_sql("  from ");
+                out.push_sql(table.qualified_name.as_str());
+                out.push_sql(" e\n  where");
+                // TODO: do we need to care about it?
+                // if self.table.has_causality_region {
+                //     out.push_sql("causality_region = ");
+                //     out.push_bind_param::<Integer, _>(&self.key.causality_region)?;
+                //     out.push_sql(" and ");
+                // }
+                if table.immutable {
+                    self.imm_range.contains(&mut out)?;
+                } else {
+                    self.mut_range.contains(&mut out)?;
+                }
             }
         }
+        out.push_sql("\norder by block_number, entity, id");
 
         Ok(())
     }