perf: Add support for multi get operation for database queries

CHANGELOG.md

@@ -15,12 +15,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 - [2378](https://github.com/FuelLabs/fuel-core/pull/2378): Use cached hash of the topic instead of calculating it on each publishing gossip message.
 
+#### Breaking
+- [2396](https://github.com/FuelLabs/fuel-core/pull/2396): Return `StorageResult<Transaction>` in `OffChainDatabase::old_transaction`.
+
 ### Added
 - [2321](https://github.com/FuelLabs/fuel-core/pull/2321): New metrics for the txpool: "The size of transactions in the txpool" (`txpool_tx_size`), "The time spent by a transaction in the txpool in seconds" (`txpool_tx_time_in_txpool_seconds`), The number of transactions in the txpool (`txpool_number_of_transactions`), "The number of transactions pending verification before entering the txpool" (`txpool_number_of_transactions_pending_verification`), "The number of executable transactions in the txpool" (`txpool_number_of_executable_transactions`), "The time it took to select transactions for inclusion in a block in nanoseconds" (`txpool_select_transaction_time_nanoseconds`), The time it took to insert a transaction in the txpool in milliseconds (`txpool_insert_transaction_time_milliseconds`).
 - [2347](https://github.com/FuelLabs/fuel-core/pull/2364): Add activity concept in order to protect against infinitely increasing DA gas price scenarios
 - [2362](https://github.com/FuelLabs/fuel-core/pull/2362): Added a new request_response protocol version `/fuel/req_res/0.0.2`. In comparison with `/fuel/req/0.0.1`, which returns an empty response when a request cannot be fulfilled, this version returns more meaningful error codes. Nodes still support the version `0.0.1` of the protocol to guarantee backward compatibility with fuel-core nodes. Empty responses received from nodes using the old protocol `/fuel/req/0.0.1` are automatically converted into an error `ProtocolV1EmptyResponse` with error code 0, which is also the only error code implemented. More specific error codes will be added in the future.
 - [2386](https://github.com/FuelLabs/fuel-core/pull/2386): Add a flag to define the maximum number of file descriptors that RocksDB can use. By default it's half of the OS limit.
 - [2376](https://github.com/FuelLabs/fuel-core/pull/2376): Add a way to fetch transactions in P2P without specifying a peer.
+- [2396](https://github.com/FuelLabs/fuel-core/pull/2396): Add support for multi get operation for database queries.
 
 ## [Version 0.40.0]
 

benches/src/db_lookup_times_utils/utils.rs

@@ -90,7 +90,9 @@ fn get_block_multi_get_method(
         .ok_or(anyhow!("empty raw block"))?;
     let block: CompressedBlock = postcard::from_bytes(raw_block.as_slice())?;
     let tx_ids = block.transactions().iter();
-    let raw_txs = database.multi_get(BenchDbColumn::Transactions.id(), tx_ids)?;
+    let raw_txs: Vec<_> = database
+        .multi_get(BenchDbColumn::Transactions.id(), tx_ids)
+        .try_collect()?;
     let txs: Vec<Transaction> = raw_txs
         .iter()
         .flatten()

crates/client/src/client/schema/primitives.rs

@@ -177,7 +177,7 @@ impl LowerHex for TxPointer {
     }
 }
 
-#[derive(cynic::Scalar, Debug, Clone)]
+#[derive(cynic::Scalar, Debug, Clone, PartialEq, Eq)]
 pub struct HexString(pub Bytes);
 
 impl From<HexString> for Vec<u8> {
@@ -194,7 +194,7 @@ impl Deref for HexString {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Bytes(pub Vec<u8>);
 
 impl FromStr for Bytes {

crates/fuel-core/src/database/block.rs

@@ -7,6 +7,7 @@ use crate::{
 };
 use fuel_core_storage::{
     iter::{
+        IntoBoxedIter,
         IterDirection,
         IteratorOverTable,
     },
@@ -23,6 +24,7 @@ use fuel_core_storage::{
     Error as StorageError,
     Result as StorageResult,
     StorageAsRef,
+    StorageBatchInspect,
 };
 use fuel_core_types::{
     blockchain::{
@@ -37,7 +39,6 @@ use fuel_core_types::{
     fuel_types::BlockHeight,
 };
 use itertools::Itertools;
-use std::borrow::Cow;
 
 impl OffChainIterableKeyValueView {
     pub fn get_block_height(&self, id: &BlockId) -> StorageResult<Option<BlockHeight>> {
@@ -65,20 +66,16 @@ impl OnChainIterableKeyValueView {
     /// Retrieve the full block and all associated transactions
     pub fn get_full_block(&self, height: &BlockHeight) -> StorageResult<Option<Block>> {
         let db_block = self.storage::<FuelBlocks>().get(height)?;
+
         if let Some(block) = db_block {
-            // fetch all the transactions
-            // TODO: Use multiget when it's implemented.
-            //  https://github.com/FuelLabs/fuel-core/issues/2344
-            let txs = block
-                .transactions()
-                .iter()
-                .map(|tx_id| {
-                    self.storage::<Transactions>()
-                        .get(tx_id)
-                        .and_then(|tx| tx.ok_or(not_found!(Transactions)))
-                        .map(Cow::into_owned)
-                })
-                .try_collect()?;
+            let transaction_ids = block.transactions().iter().into_boxed();
+            let txs = <Self as StorageBatchInspect<Transactions>>::get_batch(
+                self,
+                transaction_ids,
+            )
+            .map(|res| res.and_then(|opt| opt.ok_or(not_found!(Transactions))))
+            .try_collect()?;
+
             Ok(Some(block.into_owned().uncompress(txs)))
         } else {
             Ok(None)

crates/fuel-core/src/graphql_api/database.rs

@@ -12,8 +12,6 @@ use fuel_core_storage::{
         IntoBoxedIter,
         IterDirection,
     },
-    not_found,
-    tables::Transactions,
     transactional::AtomicView,
     Error as StorageError,
     IsNotFound,
@@ -65,6 +63,7 @@ use fuel_core_types::{
 use futures::Stream;
 use std::{
     borrow::Cow,
+    collections::BTreeMap,
     sync::Arc,
 };
 
@@ -141,29 +140,57 @@ impl ReadView {
     pub fn transaction(&self, tx_id: &TxId) -> StorageResult<Transaction> {
         let result = self.on_chain.transaction(tx_id);
         if result.is_not_found() {
-            if let Some(tx) = self.off_chain.old_transaction(tx_id)? {
-                Ok(tx)
-            } else {
-                Err(not_found!(Transactions))
-            }
+            self.off_chain.old_transaction(tx_id)
         } else {
             result
         }
     }
 
-    pub async fn transactions(
+    pub async fn transactions(&self, tx_ids: &[TxId]) -> Vec<StorageResult<Transaction>> {
+        let transactions: Vec<_> = self
+            .on_chain
+            .transactions(tx_ids.iter().into_boxed())
+            .collect();
+
+        // Give a chance for other tasks to run.
+        tokio::task::yield_now().await;
+
+        if transactions.iter().any(|result| result.is_not_found()) {
+            let on_chain_results = tx_ids.iter().enumerate().zip(transactions).collect();
+
+            self.extend_with_off_chain_results(on_chain_results).await
+        } else {
+            transactions
+        }
+    }
+
+    pub async fn extend_with_off_chain_results(
         &self,
-        tx_ids: Vec<TxId>,
+        on_chain_results: BTreeMap<(usize, &TxId), StorageResult<Transaction>>,
     ) -> Vec<StorageResult<Transaction>> {
-        // TODO: Use multiget when it's implemented.
-        //  https://github.com/FuelLabs/fuel-core/issues/2344
-        let result = tx_ids
+        let off_chain_indexed_txids: Vec<_> = on_chain_results
             .iter()
-            .map(|tx_id| self.transaction(tx_id))
-            .collect::<Vec<_>>();
-        // Give a chance to other tasks to run.
+            .filter_map(|(indexed_tx_id, result)| {
+                result.is_not_found().then_some(*indexed_tx_id)
+            })
+            .collect();
+
+        let off_chain_results = off_chain_indexed_txids.iter().copied().zip(
+            self.off_chain.old_transactions(
+                off_chain_indexed_txids
+                    .iter()
+                    .map(|(_, tx_id)| *tx_id)
+                    .into_boxed(),
+            ),
+        );
+
+        let mut results = on_chain_results;
+        results.extend(off_chain_results);
+
+        // Give a chance for other tasks to run.
         tokio::task::yield_now().await;
-        result
+
+        results.into_values().collect()
     }
 
     pub fn block(&self, height: &BlockHeight) -> StorageResult<CompressedBlock> {

crates/fuel-core/src/graphql_api/ports.rs

@@ -30,12 +30,15 @@ use fuel_core_types::{
             DaBlockHeight,
         },
     },
-    entities::relayer::{
-        message::{
-            MerkleProof,
-            Message,
+    entities::{
+        coins::coin::Coin,
+        relayer::{
+            message::{
+                MerkleProof,
+                Message,
+            },
+            transaction::RelayedTransactionStatus,
         },
-        transaction::RelayedTransactionStatus,
     },
     fuel_tx::{
         Bytes32,
@@ -104,7 +107,12 @@ pub trait OffChainDatabase: Send + Sync {
 
     fn old_block_consensus(&self, height: &BlockHeight) -> StorageResult<Consensus>;
 
-    fn old_transaction(&self, id: &TxId) -> StorageResult<Option<Transaction>>;
+    fn old_transaction(&self, id: &TxId) -> StorageResult<Transaction>;
+
+    fn old_transactions<'a>(
+        &'a self,
+        ids: BoxedIter<'a, &'a TxId>,
+    ) -> BoxedIter<'a, StorageResult<Transaction>>;
 
     fn relayed_tx_status(
         &self,
@@ -120,7 +128,7 @@ pub trait OnChainDatabase:
     + Sync
     + DatabaseBlocks
     + DatabaseMessages
-    + StorageInspect<Coins, Error = StorageError>
+    + DatabaseCoins
     + StorageRead<BlobData, Error = StorageError>
     + StorageInspect<StateTransitionBytecodeVersions, Error = StorageError>
     + StorageInspect<UploadedBytecodes, Error = StorageError>
@@ -135,6 +143,12 @@ pub trait DatabaseBlocks {
     /// Get a transaction by its id.
     fn transaction(&self, tx_id: &TxId) -> StorageResult<Transaction>;
 
+    /// Get a batch of transactions by their ids.
+    fn transactions<'a>(
+        &'a self,
+        tx_ids: BoxedIter<'a, &'a TxId>,
+    ) -> BoxedIter<'a, StorageResult<Transaction>>;
+
     /// Get a block by its height.
     fn block(&self, height: &BlockHeight) -> StorageResult<CompressedBlock>;
 
@@ -166,6 +180,11 @@ pub trait DatabaseMessages: StorageInspect<Messages, Error = StorageError> {
         direction: IterDirection,
     ) -> BoxedIter<'_, StorageResult<Message>>;
 
+    fn message_batch<'a>(
+        &'a self,
+        ids: BoxedIter<'a, &'a Nonce>,
+    ) -> BoxedIter<'a, StorageResult<Message>>;
+
     fn message_exists(&self, nonce: &Nonce) -> StorageResult<bool>;
 }
 
@@ -176,6 +195,16 @@ pub trait DatabaseRelayedTransactions {
     ) -> StorageResult<Option<RelayedTransactionStatus>>;
 }
 
+/// Trait that specifies all the getters required for coins
+pub trait DatabaseCoins: StorageInspect<Coins, Error = StorageError> {
+    fn coin(&self, utxo_id: UtxoId) -> StorageResult<Coin>;
+
+    fn coins<'a>(
+        &'a self,
+        utxo_ids: BoxedIter<'a, &'a UtxoId>,
+    ) -> BoxedIter<'a, StorageResult<Coin>>;
+}
+
 /// Trait that specifies all the getters required for contract.
 pub trait DatabaseContracts:
     StorageInspect<ContractsRawCode, Error = StorageError>

crates/fuel-core/src/query/balance/asset_query.rs

@@ -170,7 +170,7 @@ impl<'a> AssetsQuery<'a> {
             })
             .try_filter_map(move |chunk| async move {
                 let chunk = database.messages(chunk).await;
-                Ok::<_, StorageError>(Some(futures::stream::iter(chunk)))
+                Ok::<_, StorageError>(Some(chunk))
             })
             .try_flatten()
             .filter(|result| {

crates/fuel-core/src/query/coin.rs

@@ -1,11 +1,11 @@
 use crate::fuel_core_graphql_api::database::ReadView;
 use fuel_core_storage::{
-    iter::IterDirection,
-    not_found,
-    tables::Coins,
+    iter::{
+        IntoBoxedIter,
+        IterDirection,
+    },
     Error as StorageError,
     Result as StorageResult,
-    StorageAsRef,
 };
 use fuel_core_types::{
     entities::coins::coin::Coin,
@@ -20,27 +20,18 @@ use futures::{
 
 impl ReadView {
     pub fn coin(&self, utxo_id: UtxoId) -> StorageResult<Coin> {
-        let coin = self
-            .on_chain
-            .as_ref()
-            .storage::<Coins>()
-            .get(&utxo_id)?
-            .ok_or(not_found!(Coins))?
-            .into_owned();
-
-        Ok(coin.uncompress(utxo_id))
+        self.on_chain.coin(utxo_id)
     }
 
     pub async fn coins(
         &self,
         utxo_ids: Vec<UtxoId>,
     ) -> impl Iterator<Item = StorageResult<Coin>> + '_ {
-        // TODO: Use multiget when it's implemented.
-        //  https://github.com/FuelLabs/fuel-core/issues/2344
-        let coins = utxo_ids.into_iter().map(|id| self.coin(id));
-        // Give a chance to other tasks to run.
+        let coins: Vec<_> = self.on_chain.coins(utxo_ids.iter().into_boxed()).collect();
+
+        // Give a chance for other tasks to run.
         tokio::task::yield_now().await;
-        coins
+        coins.into_iter()
     }
 
     pub fn owned_coins(

crates/fuel-core/src/query/message.rs

@@ -2,6 +2,7 @@ use crate::fuel_core_graphql_api::database::ReadView;
 use fuel_core_storage::{
     iter::{
         BoxedIter,
+        IntoBoxedIter,
         IterDirection,
     },
     not_found,
@@ -81,13 +82,16 @@ impl ReadView {
     pub async fn messages(
         &self,
         ids: Vec<Nonce>,
-    ) -> impl Iterator<Item = StorageResult<Message>> + '_ {
-        // TODO: Use multiget when it's implemented.
-        //  https://github.com/FuelLabs/fuel-core/issues/2344
-        let messages = ids.into_iter().map(|id| self.message(&id));
-        // Give a chance to other tasks to run.
+    ) -> impl Stream<Item = StorageResult<Message>> {
+        let messages: Vec<_> = self
+            .on_chain
+            .message_batch(ids.iter().into_boxed())
+            .collect();
+
+        // Give a chance for other tasks to run.
         tokio::task::yield_now().await;
-        messages
+
+        futures::stream::iter(messages)
     }
 
     pub fn owned_messages<'a>(
@@ -104,7 +108,7 @@ impl ReadView {
             })
             .try_filter_map(move |chunk| async move {
                 let chunk = self.messages(chunk).await;
-                Ok::<_, StorageError>(Some(futures::stream::iter(chunk)))
+                Ok::<_, StorageError>(Some(chunk))
             })
             .try_flatten()
     }

crates/fuel-core/src/query/tx.rs

@@ -50,7 +50,7 @@ impl ReadView {
             })
             .try_filter_map(move |chunk| async move {
                 let tx_ids = chunk.iter().map(|(_, tx_id)| *tx_id).collect::<Vec<_>>();
-                let txs = self.transactions(tx_ids).await;
+                let txs = self.transactions(&tx_ids).await;
                 let txs = txs
                     .into_iter()
                     .zip(chunk)

crates/fuel-core/src/schema/block.rs

@@ -144,7 +144,7 @@ impl Block {
             .filter_map(move |tx_ids: Vec<TxId>| {
                 let async_query = query.as_ref().clone();
                 async move {
-                    let txs = async_query.transactions(tx_ids.clone()).await;
+                    let txs = async_query.transactions(&tx_ids).await;
                     let txs = txs
                         .into_iter()
                         .zip(tx_ids.into_iter())

crates/fuel-core/src/schema/tx.rs

@@ -173,7 +173,7 @@ impl TxQuery {
                                 .iter()
                                 .map(|sorted| sorted.tx_id.0)
                                 .collect::<Vec<_>>();
-                            let txs = async_query.transactions(tx_ids).await;
+                            let txs = async_query.transactions(&tx_ids).await;
                             let txs = txs.into_iter().zip(chunk.into_iter()).map(
                                 |(result, sorted)| {
                                     result.map(|tx| {