Implement schema merging

core/src/subgraph/instance.rs

@@ -41,7 +41,7 @@ where
         host_metrics: Arc<HostMetrics>,
     ) -> Result<Self, Error> {
         let subgraph_id = manifest.id.clone();
-        let network = manifest.network_name()?;
+        let network = manifest.network_name();
         let templates = manifest.templates;
 
         let mut this = SubgraphInstance {

core/src/subgraph/instance_manager.rs

@@ -247,52 +247,42 @@ impl SubgraphInstanceManager {
                         "Start subgraph";
                         "data_sources" => manifest.data_sources.len()
                     );
-
-                    match manifest.network_name() {
-                        Ok(n) => {
-                            Self::start_subgraph(
-                                logger.clone(),
-                                instances.clone(),
-                                host_builder.clone(),
-                                block_stream_builder.clone(),
-                                stores
-                                    .get(&n)
-                                    .expect(&format!(
-                                        "expected store that matches subgraph network: {}",
-                                        &n
-                                    ))
-                                    .clone(),
-                                eth_adapters
-                                    .get(&n)
-                                    .expect(&format!(
-                                        "expected eth adapter that matches subgraph network: {}",
-                                        &n
-                                    ))
-                                    .clone(),
-                                manifest,
-                                metrics_registry_for_subgraph.clone(),
-                            )
-                            .map_err(|err| {
-                                error!(
-                                    logger,
-                                    "Failed to start subgraph";
-                                    "error" => format!("{}", err),
-                                    "code" => LogCode::SubgraphStartFailure
-                                )
-                            })
-                            .and_then(|_| {
-                                manager_metrics.subgraph_count.inc();
-                                Ok(())
-                            })
-                            .ok();
-                        }
-                        Err(err) => error!(
+                    let network = manifest.network_name();
+                    Self::start_subgraph(
+                        logger.clone(),
+                        instances.clone(),
+                        host_builder.clone(),
+                        block_stream_builder.clone(),
+                        stores
+                            .get(&network)
+                            .expect(&format!(
+                                "expected store that matches subgraph network: {}",
+                                &network
+                            ))
+                            .clone(),
+                        eth_adapters
+                            .get(&network)
+                            .expect(&format!(
+                                "expected eth adapter that matches subgraph network: {}",
+                                &network
+                            ))
+                            .clone(),
+                        manifest,
+                        metrics_registry_for_subgraph.clone(),
+                    )
+                    .map_err(|err| {
+                        error!(
                             logger,
                             "Failed to start subgraph";
-                             "error" => format!("{}", err),
+                            "error" => format!("{}", err),
                             "code" => LogCode::SubgraphStartFailure
-                        ),
-                    };
+                        )
+                    })
+                    .and_then(|_| {
+                        manager_metrics.subgraph_count.inc();
+                        Ok(())
+                    })
+                    .ok();
                 }
                 SubgraphStop(id) => {
                     let logger = logger_factory.subgraph_logger(&id);
@@ -338,7 +328,7 @@ impl SubgraphInstanceManager {
 
         // Clone the deployment ID for later
         let deployment_id = manifest.id.clone();
-        let network_name = manifest.network_name()?;
+        let network_name = manifest.network_name();
 
         // Obtain filters from the manifest
         let log_filter = EthereumLogFilter::from_data_sources(&manifest.data_sources);

core/src/subgraph/mod.rs

@@ -3,7 +3,6 @@ mod instance_manager;
 mod loader;
 mod provider;
 mod registrar;
-mod validation;
 
 pub use self::instance::SubgraphInstance;
 pub use self::instance_manager::SubgraphInstanceManager;

core/src/subgraph/registrar.rs

@@ -13,7 +13,6 @@ lazy_static! {
     );
 }
 
-use super::validation;
 use graph::data::subgraph::schema::{
     generate_entity_id, SubgraphDeploymentAssignmentEntity, SubgraphDeploymentEntity,
     SubgraphEntity, SubgraphVersionEntity, TypedEntity,
@@ -40,7 +39,7 @@ impl<L, P, S, CS> SubgraphRegistrar<L, P, S, CS>
 where
     L: LinkResolver + Clone,
     P: SubgraphAssignmentProviderTrait,
-    S: Store,
+    S: Store + SubgraphDeploymentStore,
     CS: ChainStore,
 {
     pub fn new(
@@ -275,7 +274,7 @@ impl<L, P, S, CS> SubgraphRegistrarTrait for SubgraphRegistrar<L, P, S, CS>
 where
     L: LinkResolver,
     P: SubgraphAssignmentProviderTrait,
-    S: Store,
+    S: Store + SubgraphDeploymentStore,
     CS: ChainStore,
 {
     fn create_subgraph(
@@ -296,70 +295,81 @@ where
         hash: SubgraphDeploymentId,
         node_id: NodeId,
     ) -> Box<dyn Future<Item = (), Error = SubgraphRegistrarError> + Send + 'static> {
-        let store = self.store.clone();
+        let store_for_validation = self.store.clone();
+        let store_for_subgraph_version = self.store.clone();
         let chain_stores = self.chain_stores.clone();
         let ethereum_adapters = self.ethereum_adapters.clone();
         let version_switching_mode = self.version_switching_mode;
 
         let logger = self.logger_factory.subgraph_logger(&hash);
-        let logger2 = logger.clone();
-        let logger3 = logger.clone();
+        let logger_for_subgraph_version = logger.clone();
+        let logger_for_debug = logger.clone();
         let name_inner = name.clone();
 
         Box::new(
-            SubgraphManifest::resolve(hash.to_ipfs_link(), self.resolver.clone(), logger.clone())
-                .map_err(SubgraphRegistrarError::ResolveError)
-                .and_then(validation::validate_manifest)
-                .and_then(move |manifest| {
-                    manifest
-                        .network_name()
-                        .map_err(|e| SubgraphRegistrarError::ManifestValidationError(vec![e]))
-                        .and_then(move |network_name| {
-                            chain_stores
-                                .clone()
-                                .get(&network_name)
-                                .ok_or(SubgraphRegistrarError::NetworkNotSupported(
-                                    network_name.clone(),
-                                ))
-                                .and_then(move |chain_store| {
-                                    ethereum_adapters
-                                        .get(&network_name)
-                                        .ok_or(SubgraphRegistrarError::NetworkNotSupported(
-                                            network_name.clone(),
-                                        ))
-                                        .map(move |ethereum_adapter| {
-                                            (
-                                                manifest,
-                                                ethereum_adapter.clone(),
-                                                chain_store.clone(),
-                                            )
-                                        })
-                                })
-                        })
-                })
-                .and_then(move |(manifest, ethereum_adapter, chain_store)| {
+            UnvalidatedSubgraphManifest::resolve(
+                hash.to_ipfs_link(),
+                self.resolver.clone(),
+                logger,
+            )
+            .map_err(SubgraphRegistrarError::ResolveError)
+            .and_then(move |unvalidated| {
+                future::result(unvalidated.validate(store_for_validation)).map_err(
+                    |validation_errors| {
+                        SubgraphRegistrarError::ManifestValidationError(validation_errors)
+                    },
+                )
+            })
+            .and_then(move |(manifest, validation_warnings)| {
+                let network_name = manifest.network_name();
+                chain_stores
+                    .clone()
+                    .get(&network_name)
+                    .ok_or(SubgraphRegistrarError::NetworkNotSupported(
+                        network_name.clone(),
+                    ))
+                    .and_then(move |chain_store| {
+                        ethereum_adapters
+                            .get(&network_name)
+                            .ok_or(SubgraphRegistrarError::NetworkNotSupported(
+                                network_name.clone(),
+                            ))
+                            .map(move |ethereum_adapter| {
+                                (
+                                    manifest,
+                                    ethereum_adapter.clone(),
+                                    chain_store.clone(),
+                                    validation_warnings,
+                                )
+                            })
+                    })
+            })
+            .and_then(
+                move |(manifest, ethereum_adapter, chain_store, validation_warnings)| {
                     let manifest_id = manifest.id.clone();
                     create_subgraph_version(
-                        &logger2,
-                        store,
+                        &logger_for_subgraph_version,
+                        store_for_subgraph_version,
                         chain_store.clone(),
                         ethereum_adapter.clone(),
                         name,
                         manifest,
                         node_id,
                         version_switching_mode,
                     )
-                    .map(|_| manifest_id)
-                })
-                .and_then(move |manifest_id| {
-                    debug!(
-                        logger3,
-                        "Wrote new subgraph version to store";
-                        "subgraph_name" => name_inner.to_string(),
-                        "subgraph_hash" => manifest_id.to_string(),
-                    );
-                    Ok(())
-                }),
+                    .map(|_| (manifest_id, validation_warnings))
+                },
+            )
+            .and_then(move |(manifest_id, validation_warnings)| {
+                debug!(
+                    logger_for_debug,
+                    "Wrote new subgraph version to store";
+                    "subgraph_name" => name_inner.to_string(),
+                    "subgraph_hash" => manifest_id.to_string(),
+                    "validation_warnings" => format!("{:?}", validation_warnings),
+                );
+                Ok(())
+            }),
         )
     }
 

graph/src/components/store.rs

@@ -14,6 +14,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 use web3::types::H256;
 
+use crate::data::schema::{SchemaImportError, SchemaReference};
 use crate::data::store::*;
 use crate::data::subgraph::schema::*;
 use crate::prelude::*;
@@ -1187,6 +1188,19 @@ pub trait SubgraphDeploymentStore: Send + Sync + 'static {
     /// store internals that should really be hidden and should be used
     /// sparingly and only when absolutely needed
     fn uses_relational_schema(&self, subgraph_id: &SubgraphDeploymentId) -> Result<bool, Error>;
+
+    fn resolve_schema_reference(
+        &self,
+        schema_reference: &SchemaReference,
+    ) -> Result<Arc<Schema>, SchemaImportError>;
+
+    fn resolve_import_graph(
+        &self,
+        schema: &Schema,
+    ) -> (
+        HashMap<SchemaReference, Arc<Schema>>,
+        Vec<SchemaImportError>,
+    );
 }
 
 /// Common trait for blockchain store implementations.