Subgraph Composition: Support multiple subgraph datasources

chain/arweave/src/chain.rs

@@ -27,7 +27,7 @@ use graph::{
     prelude::{async_trait, o, BlockNumber, ChainStore, Error, Logger, LoggerFactory},
 };
 use prost::Message;
-use std::collections::HashSet;
+use std::collections::BTreeSet;
 use std::sync::Arc;
 
 use crate::adapter::TriggerFilter;
@@ -272,7 +272,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
     async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
-        _block_numbers: HashSet<BlockNumber>,
+        _block_numbers: BTreeSet<BlockNumber>,
     ) -> Result<Vec<ArweaveBlock>, Error> {
         todo!()
     }

chain/cosmos/src/chain.rs

@@ -4,7 +4,7 @@ use graph::components::adapter::ChainId;
 use graph::env::EnvVars;
 use graph::prelude::MetricsRegistry;
 use graph::substreams::Clock;
-use std::collections::HashSet;
+use std::collections::BTreeSet;
 use std::convert::TryFrom;
 use std::sync::Arc;
 
@@ -200,7 +200,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
     async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
-        _block_numbers: HashSet<BlockNumber>,
+        _block_numbers: BTreeSet<BlockNumber>,
     ) -> Result<Vec<Block>, Error> {
         todo!()
     }

chain/ethereum/src/chain.rs

@@ -39,7 +39,7 @@ use graph::{
     },
 };
 use prost::Message;
-use std::collections::HashSet;
+use std::collections::{BTreeSet, HashSet};
 use std::future::Future;
 use std::iter::FromIterator;
 use std::sync::Arc;
@@ -747,7 +747,7 @@ pub struct TriggersAdapter {
 async fn fetch_unique_blocks_from_cache(
     logger: &Logger,
     chain_store: Arc<dyn ChainStore>,
-    block_numbers: HashSet<BlockNumber>,
+    block_numbers: BTreeSet<BlockNumber>,
 ) -> (Vec<Arc<ExtendedBlockPtr>>, Vec<i32>) {
     // Load blocks from the cache
     let blocks_map = chain_store
@@ -795,7 +795,7 @@ async fn fetch_unique_blocks_from_cache(
 async fn load_blocks<F, Fut>(
     logger: &Logger,
     chain_store: Arc<dyn ChainStore>,
-    block_numbers: HashSet<BlockNumber>,
+    block_numbers: BTreeSet<BlockNumber>,
     fetch_missing: F,
 ) -> Result<Vec<BlockFinality>>
 where
@@ -843,7 +843,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
     async fn load_block_ptrs_by_numbers(
         &self,
         logger: Logger,
-        block_numbers: HashSet<BlockNumber>,
+        block_numbers: BTreeSet<BlockNumber>,
     ) -> Result<Vec<BlockFinality>> {
         match &*self.chain_client {
             ChainClient::Firehose(endpoints) => {
@@ -1200,7 +1200,6 @@ mod tests {
     use graph::{slog, tokio};
 
     use super::*;
-    use std::collections::HashSet;
     use std::sync::Arc;
 
     // Helper function to create test blocks
@@ -1224,7 +1223,7 @@ mod tests {
         let block = create_test_block(1, "block1");
         chain_store.blocks.insert(1, vec![block.clone()]);
 
-        let block_numbers: HashSet<_> = vec![1].into_iter().collect();
+        let block_numbers: BTreeSet<_> = vec![1].into_iter().collect();
 
         let (blocks, missing) =
             fetch_unique_blocks_from_cache(&logger, Arc::new(chain_store), block_numbers).await;
@@ -1246,7 +1245,7 @@ mod tests {
             .blocks
             .insert(1, vec![block1.clone(), block2.clone()]);
 
-        let block_numbers: HashSet<_> = vec![1].into_iter().collect();
+        let block_numbers: BTreeSet<_> = vec![1].into_iter().collect();
 
         let (blocks, missing) =
             fetch_unique_blocks_from_cache(&logger, Arc::new(chain_store), block_numbers).await;
@@ -1266,7 +1265,7 @@ mod tests {
         let block = create_test_block(1, "block1");
         chain_store.blocks.insert(1, vec![block.clone()]);
 
-        let block_numbers: HashSet<_> = vec![1, 2].into_iter().collect();
+        let block_numbers: BTreeSet<_> = vec![1, 2].into_iter().collect();
 
         let (blocks, missing) =
             fetch_unique_blocks_from_cache(&logger, Arc::new(chain_store), block_numbers).await;
@@ -1287,7 +1286,7 @@ mod tests {
         chain_store.blocks.insert(1, vec![block1.clone()]);
         chain_store.blocks.insert(2, vec![block2.clone()]);
 
-        let block_numbers: HashSet<_> = vec![1, 2].into_iter().collect();
+        let block_numbers: BTreeSet<_> = vec![1, 2].into_iter().collect();
 
         let (blocks, missing) =
             fetch_unique_blocks_from_cache(&logger, Arc::new(chain_store), block_numbers).await;
@@ -1316,7 +1315,7 @@ mod tests {
             .blocks
             .insert(2, vec![block2a.clone(), block2b.clone()]);
 
-        let block_numbers: HashSet<_> = vec![1, 2, 3].into_iter().collect();
+        let block_numbers: BTreeSet<_> = vec![1, 2, 3].into_iter().collect();
 
         let (blocks, missing) =
             fetch_unique_blocks_from_cache(&logger, Arc::new(chain_store), block_numbers).await;

chain/near/src/chain.rs

@@ -32,7 +32,7 @@ use graph::{
     prelude::{async_trait, o, BlockNumber, ChainStore, Error, Logger, LoggerFactory},
 };
 use prost::Message;
-use std::collections::HashSet;
+use std::collections::BTreeSet;
 use std::sync::Arc;
 
 use crate::adapter::TriggerFilter;
@@ -328,7 +328,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
     async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
-        _block_numbers: HashSet<BlockNumber>,
+        _block_numbers: BTreeSet<BlockNumber>,
     ) -> Result<Vec<Block>> {
         unimplemented!()
     }

chain/starknet/src/chain.rs

@@ -30,7 +30,7 @@ use graph::{
     slog::o,
 };
 use prost::Message;
-use std::{collections::HashSet, sync::Arc};
+use std::{collections::BTreeSet, sync::Arc};
 
 use crate::{
     adapter::TriggerFilter,
@@ -375,7 +375,7 @@ impl TriggersAdapterTrait<Chain> for TriggersAdapter {
     async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
-        _block_numbers: HashSet<BlockNumber>,
+        _block_numbers: BTreeSet<BlockNumber>,
     ) -> Result<Vec<StarknetBlock>, Error> {
         unimplemented!()
     }

chain/substreams/src/trigger.rs

@@ -16,7 +16,7 @@ use graph::{
 };
 use graph_runtime_wasm::module::ToAscPtr;
 use lazy_static::__Deref;
-use std::{collections::HashSet, sync::Arc};
+use std::{collections::BTreeSet, sync::Arc};
 
 use crate::{Block, Chain, NoopDataSourceTemplate, ParsedChanges};
 
@@ -139,7 +139,7 @@ impl blockchain::TriggersAdapter<Chain> for TriggersAdapter {
     async fn load_block_ptrs_by_numbers(
         &self,
         _logger: Logger,
-        _block_numbers: HashSet<BlockNumber>,
+        _block_numbers: BTreeSet<BlockNumber>,
     ) -> Result<Vec<Block>, Error> {
         unimplemented!()
     }

core/src/subgraph/runner.rs

@@ -146,6 +146,7 @@ where
                     .iter()
                     .map(|handler| handler.entity.clone())
                     .collect(),
+                manifest_idx: ds.manifest_idx,
             })
             .collect::<Vec<_>>();