remove get_range() from WritableStore

Author: Zoran Cvetkov

graph/src/components/store/mod.rs

@@ -21,7 +21,6 @@ use std::collections::btree_map::Entry;
 use std::collections::{BTreeMap, BTreeSet, HashSet};
 use std::fmt;
 use std::fmt::Display;
-use std::ops::Range;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, RwLock};
 use std::time::Duration;
@@ -1039,14 +1038,6 @@ impl ReadStore for EmptyStore {
         Ok(BTreeMap::new())
     }
 
-    fn get_range(
-        &self,
-        _entity_type: &EntityType,
-        _block_range: Range<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
-        Ok(BTreeMap::new())
-    }
-
     fn get_derived(
         &self,
         _query: &DerivedEntityQuery,

graph/src/components/store/traits.rs

@@ -193,6 +193,13 @@ pub trait SubgraphStore: Send + Sync + 'static {
         manifest_idx_and_name: Arc<Vec<(u32, String)>>,
     ) -> Result<Arc<dyn ReadStore>, StoreError>;
 
+    async fn sourceable(
+        self: Arc<Self>,
+        logger: Logger,
+        deployment: DeploymentId,
+        manifest_idx_and_name: Arc<Vec<(u32, String)>>,
+    ) -> Result<Arc<dyn SourceableStore>, StoreError>;
+
     /// Initiate a graceful shutdown of the writable that a previous call to
     /// `writable` might have started
     async fn stop_subgraph(&self, deployment: &DeploymentLocator) -> Result<(), StoreError>;
@@ -235,14 +242,6 @@ pub trait ReadStore: Send + Sync + 'static {
         keys: BTreeSet<EntityKey>,
     ) -> Result<BTreeMap<EntityKey, Entity>, StoreError>;
 
-    /// Returns all versions of entities of the given entity_type that were
-    /// changed in the given block_range.
-    fn get_range(
-        &self,
-        entity_type: &EntityType,
-        block_range: Range<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError>;
-
     /// Reverse lookup
     fn get_derived(
         &self,
@@ -265,14 +264,6 @@ impl<T: ?Sized + ReadStore> ReadStore for Arc<T> {
         (**self).get_many(keys)
     }
 
-    fn get_range(
-        &self,
-        entity_type: &EntityType,
-        block_range: Range<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
-        (**self).get_range(entity_type, block_range)
-    }
-
     fn get_derived(
         &self,
         entity_derived: &DerivedEntityQuery,
@@ -285,6 +276,27 @@ impl<T: ?Sized + ReadStore> ReadStore for Arc<T> {
     }
 }
 
+pub trait SourceableStore: Send + Sync + 'static {
+    /// Returns all versions of entities of the given entity_type that were
+    /// changed in the given block_range.
+    fn get_range(
+        &self,
+        entity_type: &EntityType,
+        block_range: Range<BlockNumber>,
+    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, 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,
+        block_range: Range<BlockNumber>,
+    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
+        (**self).get_range(entity_type, block_range)
+    }
+}
+
 pub trait DeploymentCursorTracker: Sync + Send + 'static {
     fn input_schema(&self) -> InputSchema;
 

store/postgres/src/subgraph_store.rs

@@ -44,7 +44,7 @@ use crate::{
         index::{IndexList, Method},
         Layout,
     },
-    writable::WritableStore,
+    writable::{SourceableStore, WritableStore},
     NotificationSender,
 };
 use crate::{
@@ -1548,6 +1548,19 @@ impl SubgraphStoreTrait for SubgraphStore {
             .map(|store| store as Arc<dyn store::ReadStore>)
     }
 
+    async fn sourceable(
+        self: Arc<Self>,
+        _1logger: Logger,
+        deployment: graph::components::store::DeploymentId,
+        _manifest_idx_and_name: Arc<Vec<(u32, String)>>,
+    ) -> Result<Arc<dyn store::SourceableStore>, StoreError> {
+        let deployment = deployment.into();
+        let site = self.find_site(deployment)?;
+        let store = self.for_site(&site)?;
+        let s = Arc::new(SourceableStore::new(site, store.clone()));
+        Ok(s as Arc<dyn store::SourceableStore>)
+    }
+
     async fn stop_subgraph(&self, loc: &DeploymentLocator) -> Result<(), StoreError> {
         self.evict(&loc.hash)?;
 

store/postgres/src/writable.rs

@@ -352,17 +352,6 @@ impl SyncStore {
         })
     }
 
-    fn get_range(
-        &self,
-        entity_type: &EntityType,
-        block_range: Range<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
-        retry::forever(&self.logger, "get_range", || {
-            self.writable
-                .get_range(self.site.cheap_clone(), entity_type, block_range.clone())
-        })
-    }
-
     fn get_derived(
         &self,
         key: &DerivedEntityQuery,
@@ -1568,16 +1557,6 @@ impl ReadStore for WritableStore {
         self.writer.get_many(keys)
     }
 
-    // The entities that are returned are only the ones from the database.
-    // The ones in the queue are ignored.
-    fn get_range(
-        &self,
-        entity_type: &EntityType,
-        block_range: Range<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
-        self.store.get_range(entity_type, block_range)
-    }
-
     fn get_derived(
         &self,
         key: &DerivedEntityQuery,
@@ -1590,6 +1569,28 @@ impl ReadStore for WritableStore {
     }
 }
 
+pub struct SourceableStore {
+    site: Arc<Site>,
+    store: Arc<DeploymentStore>,
+}
+
+impl SourceableStore {
+    pub fn new(site: Arc<Site>, store: Arc<DeploymentStore>) -> Self {
+        Self { site, store }
+    }
+}
+
+impl store::SourceableStore for SourceableStore {
+    fn get_range(
+        &self,
+        entity_type: &EntityType,
+        block_range: Range<BlockNumber>,
+    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
+        self.store
+            .get_range(self.site.clone(), entity_type, block_range)
+    }
+}
+
 impl DeploymentCursorTracker for WritableStore {
     fn block_ptr(&self) -> Option<BlockPtr> {
         self.block_ptr.lock().unwrap().clone()

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

@@ -20,7 +20,6 @@ use lazy_static::lazy_static;
 use slog::Logger;
 use std::collections::{BTreeMap, BTreeSet};
 use std::marker::PhantomData;
-use std::ops::Range;
 use std::sync::Arc;
 use web3::types::H256;
 
@@ -67,14 +66,6 @@ impl ReadStore for MockStore {
         Ok(self.get_many_res.clone())
     }
 
-    fn get_range(
-        &self,
-        _entity_type: &EntityType,
-        _block_range: Range<BlockNumber>,
-    ) -> Result<BTreeMap<BlockNumber, Vec<Entity>>, StoreError> {
-        todo!()
-    }
-
     fn get_derived(
         &self,
         _key: &DerivedEntityQuery,

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

@@ -9,7 +9,9 @@ use std::marker::PhantomData;
 use std::ops::Range;
 use test_store::*;
 
-use graph::components::store::{DeploymentLocator, DerivedEntityQuery, WritableStore};
+use graph::components::store::{
+    DeploymentLocator, DerivedEntityQuery, SourceableStore, WritableStore,
+};
 use graph::data::subgraph::*;
 use graph::semver::Version;
 use graph::{entity, prelude::*};
@@ -87,7 +89,14 @@ fn remove_test_data(store: Arc<DieselSubgraphStore>) {
 /// Test harness for running database integration tests.
 fn run_test<R, F>(test: F)
 where
-    F: FnOnce(Arc<DieselStore>, Arc<dyn WritableStore>, DeploymentLocator) -> R + Send + 'static,
+    F: FnOnce(
+            Arc<DieselStore>,
+            Arc<dyn WritableStore>,
+            Arc<dyn SourceableStore>,
+            DeploymentLocator,
+        ) -> R
+        + Send
+        + 'static,
     R: std::future::Future<Output = ()> + Send + 'static,
 {
     run_test_sequentially(|store| async move {
@@ -102,10 +111,15 @@ where
             .writable(LOGGER.clone(), deployment.id, Arc::new(Vec::new()))
             .await
             .expect("we can get a writable store");
+        let sourceable = store
+            .subgraph_store()
+            .sourceable(LOGGER.clone(), deployment.id, Arc::new(Vec::new()))
+            .await
+            .expect("we can get a writable store");
 
         // Run test and wait for the background writer to finish its work so
         // it won't conflict with the next test
-        test(store, writable, deployment).await;
+        test(store, writable, sourceable, deployment).await;
     });
 }
 
@@ -196,7 +210,7 @@ fn get_with_pending<F>(batch: bool, read_count: F)
 where
     F: Send + Fn(&dyn WritableStore) -> i32 + Sync + 'static,
 {
-    run_test(move |store, writable, deployment| async move {
+    run_test(move |store, writable, _, deployment| async move {
         let subgraph_store = store.subgraph_store();
 
         let read_count = || read_count(writable.as_ref());
@@ -294,7 +308,7 @@ fn get_derived_nobatch() {
 
 #[test]
 fn restart() {
-    run_test(|store, writable, deployment| async move {
+    run_test(|store, writable, _, deployment| async move {
         let subgraph_store = store.subgraph_store();
         let schema = subgraph_store.input_schema(&deployment.hash).unwrap();
 
@@ -346,6 +360,7 @@ fn restart() {
 async fn read_range(
     store: Arc<Store>,
     writable: Arc<dyn WritableStore>,
+    sourceable: Arc<dyn SourceableStore>,
     deployment: DeploymentLocator,
     mutable: bool,
 ) -> usize {
@@ -367,22 +382,26 @@ async fn read_range(
     } else {
         &COUNTER2_TYPE
     };
-    let e = writable.get_range(et, br).unwrap();
+    let e = sourceable.get_range(et, br).unwrap();
     e.iter().map(|(_, v)| v.iter()).flatten().count()
 }
 
 #[test]
 fn read_range_mutable() {
-    run_test(|store, writable, deployment| async move {
-        let num_entities = read_range(store, writable, deployment, true).await;
-        assert_eq!(num_entities, 6) // TODO: fix it - it should be 4 as the range is open
-    })
+    run_test(
+        |store, writable, sourceable: Arc<dyn SourceableStore>, deployment| async move {
+            let num_entities = read_range(store, writable, sourceable, deployment, true).await;
+            assert_eq!(num_entities, 6) // TODO: fix it - it should be 4 as the range is open
+        },
+    )
 }
 
 #[test]
 fn read_range_immutable() {
-    run_test(|store, writable, deployment| async move {
-        let num_entities = read_range(store, writable, deployment, false).await;
-        assert_eq!(num_entities, 6) // TODO: fix it - it should be 4 as the range is open
-    })
+    run_test(
+        |store, writable, sourceable: Arc<dyn SourceableStore>, deployment| async move {
+            let num_entities = read_range(store, writable, sourceable, deployment, false).await;
+            assert_eq!(num_entities, 6) // TODO: fix it - it should be 4 as the range is open
+        },
+    )
 }