Add casuality region to the SQL querry

Author: Zoran Cvetkov

graph/src/blockchain/block_stream.rs

@@ -1,5 +1,5 @@
 use crate::blockchain::SubgraphFilter;
-use crate::data_source::subgraph;
+use crate::data_source::{subgraph, CausalityRegion};
 use crate::substreams::Clock;
 use crate::substreams_rpc::response::Message as SubstreamsMessage;
 use crate::substreams_rpc::BlockScopedData;
@@ -456,7 +456,7 @@ async fn get_entities_for_range(
         let entity_type = schema.entity_type(entity_name)?;
         entity_types.push(entity_type);
     }
-    Ok(store.get_range(entity_types, from..to)?)
+    Ok(store.get_range(entity_types, CausalityRegion::ONCHAIN, from..to)?)
 }
 
 impl<C: Blockchain> TriggersAdapterWrapper<C> {

store/postgres/src/deployment_store.rs

@@ -1060,11 +1060,12 @@ impl DeploymentStore {
         &self,
         site: Arc<Site>,
         entity_types: Vec<EntityType>,
+        causality_region: CausalityRegion,
         block_range: Range<BlockNumber>,
     ) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
         let mut conn = self.get_conn()?;
         let layout = self.layout(&mut conn, site)?;
-        layout.find_range(&mut conn, entity_types, block_range)
+        layout.find_range(&mut conn, entity_types, causality_region, block_range)
     }
 
     pub(crate) fn get_derived(

store/postgres/src/relational.rs

@@ -520,6 +520,7 @@ impl Layout {
         &self,
         conn: &mut PgConnection,
         entity_types: Vec<EntityType>,
+        causality_region: CausalityRegion,
         block_range: Range<BlockNumber>,
     ) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
         let mut tables = vec![];
@@ -529,11 +530,11 @@ impl Layout {
             et_map.insert(et.to_string(), Arc::new(et));
         }
         let mut entities: BTreeMap<BlockNumber, Vec<EntityWithType>> = BTreeMap::new();
-        let lower_vec = FindRangeQuery::new(&tables, false, block_range.clone())
+        let lower_vec = FindRangeQuery::new(&tables, causality_region, false, block_range.clone())
             .get_results::<EntityDataExt>(conn)
             .optional()?
             .unwrap_or_default();
-        let upper_vec = FindRangeQuery::new(&tables, true, block_range)
+        let upper_vec = FindRangeQuery::new(&tables, causality_region, true, block_range)
             .get_results::<EntityDataExt>(conn)
             .optional()?
             .unwrap_or_default();

store/postgres/src/relational_queries.rs

@@ -2024,6 +2024,7 @@ impl<'a, Conn> RunQueryDsl<Conn> for FindQuery<'a> {}
 #[derive(Debug, Clone)]
 pub struct FindRangeQuery<'a> {
     tables: &'a Vec<&'a Table>,
+    causality_region: CausalityRegion,
     is_upper_range: bool,
     imm_range: EntityBlockRange,
     mut_range: EntityBlockRange,
@@ -2032,13 +2033,15 @@ pub struct FindRangeQuery<'a> {
 impl<'a> FindRangeQuery<'a> {
     pub fn new(
         tables: &'a Vec<&Table>,
+        causality_region: CausalityRegion,
         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,
+            causality_region,
             is_upper_range,
             imm_range,
             mut_range,
@@ -2075,12 +2078,12 @@ impl<'a> QueryFragment<Pg> for FindRangeQuery<'a> {
                 out.push_sql("  from ");
                 out.push_sql(table.qualified_name.as_str());
                 out.push_sql(" e\n  where");
-                // TODO: add casuality region to the query
-                // if self.table.has_causality_region {
-                //     out.push_sql("causality_region = ");
-                //     out.push_bind_param::<Integer, _>(&self.key.causality_region)?;
-                //     out.push_sql(" and ");
-                // }
+                // add casuality region to the query
+                if table.has_causality_region {
+                    out.push_sql("causality_region = ");
+                    out.push_bind_param::<Integer, _>(&self.causality_region)?;
+                    out.push_sql(" and ");
+                }
                 if table.immutable {
                     self.imm_range.contains(&mut out)?;
                 } else {

store/test-store/tests/postgres/writable.rs

@@ -376,58 +376,30 @@ fn read_range_test() {
             let subgraph_store = store.subgraph_store();
             writable.deployment_synced().unwrap();
 
-            // First test sourceable store with individual types
-            let mutable_count = read_range(
-                store.clone(),
-                writable.clone(),
-                sourceable.clone(),
-                deployment.clone(),
-                true,
-            )
-            .await;
-            let immutable_count = read_range(
-                store.clone(),
-                writable.clone(),
-                sourceable.clone(),
-                deployment.clone(),
-                false,
-            )
-            .await;
-            assert_eq!(mutable_count, 4); // Fixed: range is half-open
-            assert_eq!(immutable_count, 4); // Fixed: range is half-open
-
-            // Then test writable store with the detailed entity checks
-            for count in 1..=5 {
-                insert_count(&subgraph_store, &deployment, count, 2 * count, true).await;
-            }
-            writable.flush().await.unwrap();
-            writable.deployment_synced().unwrap();
-
-            let br: Range<BlockNumber> = 0..18;
-            let entity_types = vec![COUNTER_TYPE.clone(), COUNTER2_TYPE.clone()];
-            let e: BTreeMap<i32, Vec<EntityWithType>> = writable
-                .get_range(entity_types.clone(), br.clone())
-                .unwrap();
-            assert_eq!(e.len(), 5);
-            for en in &e {
-                let index = *en.0 - 1;
-                let a = result_entities[index as usize];
-                assert_eq!(a, format!("{:?}", en));
-            }
-
-            for count in 6..=7 {
-                insert_count(&subgraph_store, &deployment, count, 2 * count, true).await;
-            }
-            writable.flush().await.unwrap();
-            writable.deployment_synced().unwrap();
-
-            let e: BTreeMap<i32, Vec<EntityWithType>> = writable.get_range(entity_types, br).unwrap();
-            assert_eq!(e.len(), 7);
-            for en in &e {
-                let index = *en.0 - 1;
-                let a = result_entities[index as usize];
-                assert_eq!(a, format!("{:?}", en));
-            }
-        },
-    )
-}
+        let br: Range<BlockNumber> = 0..18;
+        let entity_types = vec![COUNTER_TYPE.clone(), COUNTER2_TYPE.clone()];
+        let e: BTreeMap<i32, Vec<EntityWithType>> = writable
+            .get_range(entity_types.clone(), CausalityRegion::ONCHAIN, br.clone())
+            .unwrap();
+        assert_eq!(e.len(), 5);
+        for en in &e {
+            let index = *en.0 - 1;
+            let a = result_entities[index as usize];
+            assert_eq!(a, format!("{:?}", en));
+        }
+        for count in 6..=7 {
+            insert_count(&subgraph_store, &deployment, count, 2 * count, true).await;
+        }
+        writable.flush().await.unwrap();
+        writable.deployment_synced().unwrap();
+        let e: BTreeMap<i32, Vec<EntityWithType>> = writable
+            .get_range(entity_types, CausalityRegion::ONCHAIN, br)
+            .unwrap();
+        assert_eq!(e.len(), 7);
+        for en in &e {
+            let index = *en.0 - 1;
+            let a = result_entities[index as usize];
+            assert_eq!(a, format!("{:?}", en));
+        }
+    })
+}