fixes

Author: Zoran Cvetkov

graph/src/components/store/mod.rs

@@ -25,7 +25,6 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, RwLock};
 use std::time::Duration;
 
-use crate::blockchain::block_stream::EntityWithType;
 use crate::blockchain::{Block, BlockHash, BlockPtr};
 use crate::cheap_clone::CheapClone;
 use crate::components::store::write::EntityModification;

graph/src/components/store/traits.rs

@@ -298,19 +298,21 @@ pub trait SourceableStore: DeploymentCursorTracker {
     /// changed in the given block_range.
     fn get_range(
         &self,
-        entity_type: &EntityType,
+        entity_types: Vec<EntityType>,
+        causality_region: CausalityRegion,
         block_range: Range<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError>;
+    ) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError>;
 }
 
 // This silly impl is needed until https://github.com/rust-lang/rust/issues/65991 is stable.
 impl<T: ?Sized + SourceableStore> SourceableStore for Arc<T> {
     fn get_range(
         &self,
-        entity_type: &EntityType,
+        entity_types: Vec<EntityType>,
+        causality_region: CausalityRegion,
         block_range: Range<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
-        (**self).get_range(entity_type, block_range)
+    ) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
+        (**self).get_range(entity_types, causality_region, block_range)
     }
 }
 

store/postgres/src/block_range.rs

@@ -148,7 +148,7 @@ impl EntityBlockRange {
         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 table.immutable {
+        if immutable {
             Self::Immutable(block_range)
         } else {
             Self::Mutable((block_range, is_uppper_range))

store/postgres/src/writable.rs

@@ -1598,11 +1598,16 @@ impl SourceableStore {
 impl store::SourceableStore for SourceableStore {
     fn get_range(
         &self,
-        entity_type: &EntityType,
+        entity_types: Vec<EntityType>,
+        causality_region: CausalityRegion,
         block_range: Range<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
-        self.store
-            .get_range(self.site.clone(), entity_type, block_range)
+    ) -> Result<BTreeMap<BlockNumber, Vec<EntityWithType>>, StoreError> {
+        self.store.get_range(
+            self.site.clone(),
+            entity_types,
+            causality_region,
+            block_range,
+        )
     }
 }
 impl DeploymentCursorTracker for SourceableStore {

store/test-store/tests/graph/entity_cache.rs

@@ -1,4 +1,4 @@
-use graph::blockchain::block_stream::{EntityWithType, FirehoseCursor};
+use graph::blockchain::block_stream::FirehoseCursor;
 use graph::blockchain::BlockTime;
 use graph::components::store::{
     DeploymentCursorTracker, DerivedEntityQuery, GetScope, LoadRelatedRequest, ReadStore,

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

@@ -333,52 +333,30 @@ fn restart() {
     })
 }
 
-async fn read_range(
-    store: Arc<Store>,
-    writable: Arc<dyn WritableStore>,
-    sourceable: Arc<dyn SourceableStore>,
-    deployment: DeploymentLocator,
-    mutable: bool,
-) -> usize {
-    let subgraph_store = store.subgraph_store();
-    writable.deployment_synced().unwrap();
-
-    for count in 1..=7 {
-        insert_count_mutable(&subgraph_store, &deployment, 2 * count, 4 * count).await;
-        insert_count_immutable(&subgraph_store, &deployment, 2 * count + 1, 4 * count).await;
-    }
-    writable.flush().await.unwrap();
-
-    let br: Range<BlockNumber> = 4..8;
-    let et: &EntityType = if mutable {
-        &COUNTER_TYPE
-    } else {
-        &COUNTER2_TYPE
-    };
-    let e = sourceable.get_range(et, br).unwrap();
-    e.iter().map(|(_, v)| v.iter()).flatten().count()
-}
-
 #[test]
 fn read_range_test() {
-    let result_entities = vec![
-        r#"(1, [EntityWithType { entity_op: Create, entity_type: EntityType(Counter), entity: Entity { count: Int(2), id: String("1") }, vid: 1 }, EntityWithType { entity_op: Create, entity_type: EntityType(Counter2), entity: Entity { count: Int(2), id: String("1") }, vid: 1 }])"#,
-        r#"(2, [EntityWithType { entity_op: Modify, entity_type: EntityType(Counter), entity: Entity { count: Int(4), id: String("1") }, vid: 2 }, EntityWithType { entity_op: Create, entity_type: EntityType(Counter2), entity: Entity { count: Int(4), id: String("2") }, vid: 2 }])"#,
-        r#"(3, [EntityWithType { entity_op: Delete, entity_type: EntityType(Counter), entity: Entity { count: Int(4), id: String("1") }, vid: 2 }, EntityWithType { entity_op: Create, entity_type: EntityType(Counter2), entity: Entity { count: Int(6), id: String("3") }, vid: 3 }])"#,
-        r#"(4, [EntityWithType { entity_op: Create, entity_type: EntityType(Counter), entity: Entity { count: Int(8), id: String("1") }, vid: 3 }, EntityWithType { entity_op: Create, entity_type: EntityType(Counter2), entity: Entity { count: Int(8), id: String("4") }, vid: 4 }])"#,
-        r#"(5, [EntityWithType { entity_op: Delete, entity_type: EntityType(Counter), entity: Entity { count: Int(8), id: String("1") }, vid: 3 }, EntityWithType { entity_op: Create, entity_type: EntityType(Counter2), entity: Entity { count: Int(10), id: String("5") }, vid: 5 }])"#,
-        r#"(6, [EntityWithType { entity_op: Create, entity_type: EntityType(Counter), entity: Entity { count: Int(12), id: String("1") }, vid: 4 }])"#,
-        r#"(7, [EntityWithType { entity_op: Delete, entity_type: EntityType(Counter), entity: Entity { count: Int(12), id: String("1") }, vid: 4 }])"#,
-    ];
-
-    run_test(
-        |store, writable, sourceable: Arc<dyn SourceableStore>, deployment| async move {
-            let subgraph_store = store.subgraph_store();
-            writable.deployment_synced().unwrap();
+    run_test(|store, writable, sourceable, deployment| async move {
+        let result_entities = vec![
+            r#"(1, [EntityWithType { entity_op: Create, entity_type: EntityType(Counter), entity: Entity { count: Int(2), id: String("1") }, vid: 1 }, EntityWithType { entity_op: Create, entity_type: EntityType(Counter2), entity: Entity { count: Int(2), id: String("1") }, vid: 1 }])"#,
+            r#"(2, [EntityWithType { entity_op: Modify, entity_type: EntityType(Counter), entity: Entity { count: Int(4), id: String("1") }, vid: 2 }, EntityWithType { entity_op: Create, entity_type: EntityType(Counter2), entity: Entity { count: Int(4), id: String("2") }, vid: 2 }])"#,
+            r#"(3, [EntityWithType { entity_op: Delete, entity_type: EntityType(Counter), entity: Entity { count: Int(4), id: String("1") }, vid: 2 }, EntityWithType { entity_op: Create, entity_type: EntityType(Counter2), entity: Entity { count: Int(6), id: String("3") }, vid: 3 }])"#,
+            r#"(4, [EntityWithType { entity_op: Create, entity_type: EntityType(Counter), entity: Entity { count: Int(8), id: String("1") }, vid: 3 }, EntityWithType { entity_op: Create, entity_type: EntityType(Counter2), entity: Entity { count: Int(8), id: String("4") }, vid: 4 }])"#,
+            r#"(5, [EntityWithType { entity_op: Delete, entity_type: EntityType(Counter), entity: Entity { count: Int(8), id: String("1") }, vid: 3 }, EntityWithType { entity_op: Create, entity_type: EntityType(Counter2), entity: Entity { count: Int(10), id: String("5") }, vid: 5 }])"#,
+            r#"(6, [EntityWithType { entity_op: Create, entity_type: EntityType(Counter), entity: Entity { count: Int(12), id: String("1") }, vid: 4 }])"#,
+            r#"(7, [EntityWithType { entity_op: Delete, entity_type: EntityType(Counter), entity: Entity { count: Int(12), id: String("1") }, vid: 4 }])"#,
+        ];
+        let subgraph_store = store.subgraph_store();
+        writable.deployment_synced().unwrap();
+
+        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
+        let e: BTreeMap<i32, Vec<EntityWithType>> = sourceable
             .get_range(entity_types.clone(), CausalityRegion::ONCHAIN, br.clone())
             .unwrap();
         assert_eq!(e.len(), 5);
@@ -392,7 +370,7 @@ fn read_range_test() {
         }
         writable.flush().await.unwrap();
         writable.deployment_synced().unwrap();
-        let e: BTreeMap<i32, Vec<EntityWithType>> = writable
+        let e: BTreeMap<i32, Vec<EntityWithType>> = sourceable
             .get_range(entity_types, CausalityRegion::ONCHAIN, br)
             .unwrap();
         assert_eq!(e.len(), 7);