Skip to content

optimize postgresql queries for API Server scanner #1878

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Mar 27, 2025
Merged

optimize postgresql queries for API Server scanner #1878

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
7f526f2
optimize postgresql queries for API Server scanner
OBorce Jan 20, 2025
bb3c7fb
Add delegations_balance to pool endpoints
OBorce Feb 19, 2025
9babfb2
Add token balances to the address endpoint in Api Server
OBorce Mar 11, 2025
2b01a6b
fix comments
OBorce Mar 24, 2025
c0deb6f
Add address token authority endpoint to API Server
OBorce Mar 24, 2025
51f7038
Merge pull request #1896 from mintlayer/feature/api-server-address-to…
OBorce Mar 26, 2025
c09d830
reformat code in json! macros
OBorce Mar 26, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
151 changes: 102 additions & 49 deletions api-server/api-server-common/src/storage/impls/in_memory/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,9 @@ pub mod transactional;

use crate::storage::storage_api::{
block_aux_data::{BlockAuxData, BlockWithExtraData},
ApiServerStorageError, BlockInfo, CoinOrTokenStatistic, Delegation, FungibleTokenData,
LockedUtxo, NftWithOwner, Order, PoolBlockStats, TransactionInfo, Utxo, UtxoLock,
UtxoWithExtraInfo,
AmountWithDecimals, ApiServerStorageError, BlockInfo, CoinOrTokenStatistic, Delegation,
FungibleTokenData, LockedUtxo, NftWithOwner, Order, PoolBlockStats, PoolDataWithExtraInfo,
TransactionInfo, Utxo, UtxoLock, UtxoWithExtraInfo,
};
use common::{
address::Address,
Expand All @@ -31,7 +31,6 @@ use common::{
},
primitives::{id::WithId, Amount, BlockHeight, CoinOrTokenId, Id},
};
use pos_accounting::PoolData;
use std::{
cmp::Reverse,
collections::{BTreeMap, BTreeSet},
Expand All @@ -45,24 +44,25 @@ use super::CURRENT_STORAGE_VERSION;
struct ApiServerInMemoryStorage {
block_table: BTreeMap<Id<Block>, BlockWithExtraData>,
block_aux_data_table: BTreeMap<Id<Block>, BlockAuxData>,
address_balance_table: BTreeMap<String, BTreeMap<(CoinOrTokenId, BlockHeight), Amount>>,
address_balance_table: BTreeMap<String, BTreeMap<CoinOrTokenId, BTreeMap<BlockHeight, Amount>>>,
address_locked_balance_table: BTreeMap<String, BTreeMap<(CoinOrTokenId, BlockHeight), Amount>>,
address_transactions_table: BTreeMap<String, BTreeMap<BlockHeight, Vec<Id<Transaction>>>>,
delegation_table: BTreeMap<DelegationId, BTreeMap<BlockHeight, Delegation>>,
main_chain_blocks_table: BTreeMap<BlockHeight, Id<Block>>,
pool_data_table: BTreeMap<PoolId, BTreeMap<BlockHeight, PoolData>>,
pool_data_table: BTreeMap<PoolId, BTreeMap<BlockHeight, PoolDataWithExtraInfo>>,
transaction_table: BTreeMap<Id<Transaction>, (Id<Block>, TransactionInfo)>,
utxo_table: BTreeMap<UtxoOutPoint, BTreeMap<BlockHeight, Utxo>>,
address_utxos: BTreeMap<String, BTreeSet<UtxoOutPoint>>,
locked_utxo_table: BTreeMap<UtxoOutPoint, BTreeMap<BlockHeight, LockedUtxo>>,
address_locked_utxos: BTreeMap<String, BTreeSet<UtxoOutPoint>>,
fungible_token_issuances: BTreeMap<TokenId, BTreeMap<BlockHeight, FungibleTokenData>>,
fungible_token_data: BTreeMap<TokenId, BTreeMap<BlockHeight, FungibleTokenData>>,
nft_token_issuances: BTreeMap<TokenId, BTreeMap<BlockHeight, NftWithOwner>>,
statistics:
BTreeMap<CoinOrTokenStatistic, BTreeMap<CoinOrTokenId, BTreeMap<BlockHeight, Amount>>>,
orders_table: BTreeMap<OrderId, BTreeMap<BlockHeight, Order>>,
best_block: BlockAuxData,
genesis_block: Arc<WithId<Genesis>>,
number_of_coin_decimals: u8,
storage_version: u32,
}

Expand All @@ -82,7 +82,7 @@ impl ApiServerInMemoryStorage {
address_utxos: BTreeMap::new(),
locked_utxo_table: BTreeMap::new(),
address_locked_utxos: BTreeMap::new(),
fungible_token_issuances: BTreeMap::new(),
fungible_token_data: BTreeMap::new(),
nft_token_issuances: BTreeMap::new(),
statistics: BTreeMap::new(),
orders_table: BTreeMap::new(),
Expand All @@ -92,6 +92,7 @@ impl ApiServerInMemoryStorage {
0.into(),
chain_config.genesis_block().timestamp(),
),
number_of_coin_decimals: chain_config.coin_decimals(),
storage_version: super::CURRENT_STORAGE_VERSION,
};
result
Expand All @@ -109,15 +110,46 @@ impl ApiServerInMemoryStorage {
address: &str,
coin_or_token_id: CoinOrTokenId,
) -> Result<Option<Amount>, ApiServerStorageError> {
self.address_balance_table.get(address).map_or_else(
|| Ok(None),
|balance| {
let range_begin = (coin_or_token_id, BlockHeight::zero());
let range_end = (coin_or_token_id, BlockHeight::max());
let range = balance.range(range_begin..=range_end);
Ok(range.last().map(|(_, v)| *v))
self.address_balance_table
.get(address)
.and_then(|by_coin_or_token| by_coin_or_token.get(&coin_or_token_id))
.map_or_else(
|| Ok(None),
|by_height| Ok(by_height.values().last().copied()),
)
}

fn get_address_balances(
&self,
address: &str,
) -> Result<BTreeMap<CoinOrTokenId, AmountWithDecimals>, ApiServerStorageError> {
let res = self.address_balance_table.get(address).map_or_else(
BTreeMap::new,
|by_coin_or_token| {
by_coin_or_token
.iter()
.map(|(coin_or_token_id, by_height)| {
let number_of_decimals = match coin_or_token_id {
CoinOrTokenId::Coin => self.number_of_coin_decimals,
CoinOrTokenId::TokenId(token_id) => {
self.fungible_token_data.get(token_id).map_or(0, |by_height| {
by_height.values().last().expect("not empty").number_of_decimals
})
}
};

(
*coin_or_token_id,
AmountWithDecimals {
amount: *by_height.values().last().expect("not empty"),
decimals: number_of_decimals,
},
)
})
.collect()
},
)
);
Ok(res)
}

fn get_address_locked_balance(
Expand Down Expand Up @@ -416,7 +448,7 @@ impl ApiServerInMemoryStorage {
&self,
len: u32,
offset: u32,
) -> Result<Vec<(PoolId, PoolData)>, ApiServerStorageError> {
) -> Result<Vec<(PoolId, PoolDataWithExtraInfo)>, ApiServerStorageError> {
let len = len as usize;
let offset = offset as usize;
let mut pool_data: Vec<_> = self
Expand Down Expand Up @@ -447,7 +479,7 @@ impl ApiServerInMemoryStorage {
&self,
len: u32,
offset: u32,
) -> Result<Vec<(PoolId, PoolData)>, ApiServerStorageError> {
) -> Result<Vec<(PoolId, PoolDataWithExtraInfo)>, ApiServerStorageError> {
let len = len as usize;
let offset = offset as usize;
let mut pool_data: Vec<_> = self
Expand Down Expand Up @@ -482,7 +514,10 @@ impl ApiServerInMemoryStorage {
Ok(Some(*block_id))
}

fn get_pool_data(&self, pool_id: PoolId) -> Result<Option<PoolData>, ApiServerStorageError> {
fn get_pool_data(
&self,
pool_id: PoolId,
) -> Result<Option<PoolDataWithExtraInfo>, ApiServerStorageError> {
let pool_data_result = self.pool_data_table.get(&pool_id);
match pool_data_result {
Some(data) => Ok(data.last_key_value().map(|(_, v)| v.clone())),
Expand Down Expand Up @@ -588,12 +623,29 @@ impl ApiServerInMemoryStorage {
.collect())
}

fn get_fungible_tokens_by_authority(
&self,
authority: Destination,
) -> Result<Vec<TokenId>, ApiServerStorageError> {
Ok(self
.fungible_token_data
.iter()
.filter_map(|(token_id, by_height)| {
by_height
.values()
.last()
.filter(|last| last.authority == authority)
.map(|_| *token_id)
})
.collect())
}

fn get_fungible_token_issuance(
&self,
token_id: TokenId,
) -> Result<Option<FungibleTokenData>, ApiServerStorageError> {
Ok(self
.fungible_token_issuances
.fungible_token_data
.get(&token_id)
.map(|by_height| by_height.values().last().cloned().expect("not empty")))
}
Expand All @@ -613,15 +665,15 @@ impl ApiServerInMemoryStorage {
token_id: TokenId,
) -> Result<Option<u8>, ApiServerStorageError> {
Ok(self
.fungible_token_issuances
.fungible_token_data
.get(&token_id)
.map(|data| data.values().last().expect("not empty").number_of_decimals)
.or_else(|| self.nft_token_issuances.get(&token_id).map(|_| 0)))
}

fn get_token_ids(&self, len: u32, offset: u32) -> Result<Vec<TokenId>, ApiServerStorageError> {
Ok(self
.fungible_token_issuances
.fungible_token_data
.keys()
.chain(self.nft_token_issuances.keys())
.skip(offset as usize)
Expand All @@ -637,7 +689,7 @@ impl ApiServerInMemoryStorage {
ticker: &[u8],
) -> Result<Vec<TokenId>, ApiServerStorageError> {
Ok(self
.fungible_token_issuances
.fungible_token_data
.iter()
.filter_map(|(key, value)| {
(value.values().last().expect("not empty").token_ticker == ticker).then_some(key)
Expand Down Expand Up @@ -734,7 +786,7 @@ impl ApiServerInMemoryStorage {
self.transaction_table.clear();
self.utxo_table.clear();
self.address_utxos.clear();
self.fungible_token_issuances.clear();
self.fungible_token_data.clear();
self.nft_token_issuances.clear();
self.orders_table.clear();

Expand All @@ -747,16 +799,12 @@ impl ApiServerInMemoryStorage {
) -> Result<(), ApiServerStorageError> {
// Inefficient, but acceptable for testing with InMemoryStorage

self.address_balance_table.iter_mut().for_each(|(_, balance)| {
balance
.iter()
.filter(|((_, height), _)| *height > block_height)
.map(|(key, _)| *key)
.collect::<Vec<_>>()
.iter()
.for_each(|key| {
balance.remove(key);
})
self.address_balance_table.retain(|_, by_coin_or_token| {
by_coin_or_token.retain(|_, by_block_height| {
by_block_height.retain(|height, _| height <= &block_height);
!by_block_height.is_empty()
});
!by_coin_or_token.is_empty()
});

Ok(())
Expand Down Expand Up @@ -813,7 +861,9 @@ impl ApiServerInMemoryStorage {
self.address_balance_table
.entry(address.to_string())
.or_default()
.entry((coin_or_token_id, block_height))
.entry(coin_or_token_id)
.or_default()
.entry(block_height)
.and_modify(|e| *e = amount)
.or_insert(amount);

Expand Down Expand Up @@ -963,7 +1013,7 @@ impl ApiServerInMemoryStorage {
fn set_pool_data_at_height(
&mut self,
pool_id: PoolId,
pool_data: &PoolData,
pool_data: &PoolDataWithExtraInfo,
block_height: BlockHeight,
) -> Result<(), ApiServerStorageError> {
self.pool_data_table
Expand Down Expand Up @@ -1072,16 +1122,13 @@ impl ApiServerInMemoryStorage {
Ok(())
}

fn set_fungible_token_issuance(
fn set_fungible_token_data(
&mut self,
token_id: TokenId,
block_height: BlockHeight,
issuance: FungibleTokenData,
data: FungibleTokenData,
) -> Result<(), ApiServerStorageError> {
self.fungible_token_issuances
.entry(token_id)
.or_default()
.insert(block_height, issuance);
self.fungible_token_data.entry(token_id).or_default().insert(block_height, data);
Ok(())
}

Expand All @@ -1092,21 +1139,27 @@ impl ApiServerInMemoryStorage {
issuance: NftIssuance,
owner: &Destination,
) -> Result<(), ApiServerStorageError> {
self.nft_token_issuances.entry(token_id).or_default().insert(
block_height,
NftWithOwner {
nft: issuance,
owner: Some(owner.clone()),
},
let res = self.nft_token_issuances.insert(
token_id,
BTreeMap::from([(
block_height,
NftWithOwner {
nft: issuance,
owner: Some(owner.clone()),
},
)]),
);

assert!(res.is_none(), "multiple nft issuances with same token_id");

Ok(())
}

fn del_token_issuance_above_height(
&mut self,
block_height: BlockHeight,
) -> Result<(), ApiServerStorageError> {
self.fungible_token_issuances.retain(|_, v| {
self.fungible_token_data.retain(|_, v| {
v.retain(|k, _| k <= &block_height);
!v.is_empty()
});
Expand Down
27 changes: 20 additions & 7 deletions api-server/api-server-common/src/storage/impls/in_memory/transactional/read.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,11 @@ use common::{
},
primitives::{Amount, BlockHeight, CoinOrTokenId, Id},
};
use pos_accounting::PoolData;

use crate::storage::storage_api::{
block_aux_data::BlockAuxData, ApiServerStorageError, ApiServerStorageRead, BlockInfo,
CoinOrTokenStatistic, Delegation, FungibleTokenData, NftWithOwner, Order, PoolBlockStats,
TransactionInfo, Utxo, UtxoWithExtraInfo,
block_aux_data::BlockAuxData, AmountWithDecimals, ApiServerStorageError, ApiServerStorageRead,
BlockInfo, CoinOrTokenStatistic, Delegation, FungibleTokenData, NftWithOwner, Order,
PoolBlockStats, PoolDataWithExtraInfo, TransactionInfo, Utxo, UtxoWithExtraInfo,
};

use super::ApiServerInMemoryStorageTransactionalRo;
Expand All @@ -46,6 +45,13 @@ impl ApiServerStorageRead for ApiServerInMemoryStorageTransactionalRo<'_> {
self.transaction.get_address_balance(address, coin_or_token_id)
}

async fn get_address_balances(
&self,
address: &str,
) -> Result<BTreeMap<CoinOrTokenId, AmountWithDecimals>, ApiServerStorageError> {
self.transaction.get_address_balances(address)
}

async fn get_address_locked_balance(
&self,
address: &str,
Expand Down Expand Up @@ -116,15 +122,15 @@ impl ApiServerStorageRead for ApiServerInMemoryStorageTransactionalRo<'_> {
&self,
len: u32,
offset: u32,
) -> Result<Vec<(PoolId, PoolData)>, ApiServerStorageError> {
) -> Result<Vec<(PoolId, PoolDataWithExtraInfo)>, ApiServerStorageError> {
self.transaction.get_latest_pool_ids(len, offset)
}

async fn get_pool_data_with_largest_staker_balance(
&self,
len: u32,
offset: u32,
) -> Result<Vec<(PoolId, PoolData)>, ApiServerStorageError> {
) -> Result<Vec<(PoolId, PoolDataWithExtraInfo)>, ApiServerStorageError> {
self.transaction.get_pool_data_with_largest_staker_balance(len, offset)
}

Expand Down Expand Up @@ -166,7 +172,7 @@ impl ApiServerStorageRead for ApiServerInMemoryStorageTransactionalRo<'_> {
async fn get_pool_data(
&self,
pool_id: PoolId,
) -> Result<Option<PoolData>, ApiServerStorageError> {
) -> Result<Option<PoolDataWithExtraInfo>, ApiServerStorageError> {
self.transaction.get_pool_data(pool_id)
}

Expand Down Expand Up @@ -206,6 +212,13 @@ impl ApiServerStorageRead for ApiServerInMemoryStorageTransactionalRo<'_> {
self.transaction.get_delegations_from_address(address)
}

async fn get_fungible_tokens_by_authority(
&self,
authority: Destination,
) -> Result<Vec<TokenId>, ApiServerStorageError> {
self.transaction.get_fungible_tokens_by_authority(authority)
}

async fn get_fungible_token_issuance(
&self,
token_id: TokenId,
Expand Down
Loading