fixed: add_filter_clause(), drop OR in favor of UNION clause

postgre-client/src/asset_filter_client.rs

@@ -35,7 +35,7 @@ impl PgClient {
         options: &'a GetByMethodsOptions,
     ) -> Result<(QueryBuilder<'a, Postgres>, bool), IndexDbError> {
         let mut query_builder = QueryBuilder::new(
-            "SELECT ast_pubkey pubkey, ast_slot_created slot_created, ast_slot_updated slot_updated FROM assets_v3 ",
+            "(SELECT ast_pubkey pubkey, ast_slot_created slot_created, ast_slot_updated slot_updated FROM assets_v3 ",
         );
         let group_clause_required = add_filter_clause(&mut query_builder, filter, options);
 
@@ -102,19 +102,59 @@ impl PgClient {
         }
         // Add GROUP BY clause if necessary
         if group_clause_required {
-            query_builder.push(" GROUP BY assets_v3.ast_pubkey, assets_v3.ast_slot_created, assets_v3.ast_slot_updated ");
+            query_builder.push(" GROUP BY ast_pubkey, ast_slot_created, ast_slot_updated ");
         }
+        query_builder.push(") ");
+
+        // the function with a side-effect that mutates the query_builder
+        if let Some(owner_address) = &filter.owner_address {
+            if let Some(ref token_type) = filter.token_type {
+                match *token_type {
+                    TokenType::Fungible
+                    | TokenType::NonFungible
+                    | TokenType::RegularNFT
+                    | TokenType::CompressedNFT => {},
+                    TokenType::All => {
+                        query_builder.push("  UNION ");
+                        query_builder.push("  ( ");
+                        query_builder.push(
+                            "SELECT
+                    ast_pubkey,
+                    ast_slot_created,
+                    ast_slot_updated
+                    FROM assets_v3
+                    JOIN fungible_tokens ON ast_pubkey = fungible_tokens.fbt_asset
+                    WHERE ast_supply > 0 AND fbt_owner = ",
+                        );
+                        query_builder.push_bind(owner_address);
+                        if !options.show_zero_balance {
+                            query_builder.push(" AND fbt_balance > ");
+                            query_builder.push_bind(0i64);
+                        }
+                        if !options.show_unverified_collections {
+                            // if there is no collection for asset it doesn't mean that it's unverified
+                            query_builder.push(
+                                " AND assets_v3.ast_is_collection_verified IS DISTINCT FROM FALSE",
+                            );
+                        }
+                        query_builder.push(")");
+                    },
+                }
+            }
+        }
+
         // Add ORDER BY clause
         let direction = match (&order.sort_direction, order_reversed) {
             (AssetSortDirection::Asc, true) | (AssetSortDirection::Desc, false) => " DESC ",
             (AssetSortDirection::Asc, false) | (AssetSortDirection::Desc, true) => " ASC ",
         };
 
         query_builder.push(" ORDER BY ");
-        query_builder.push(order.sort_by.to_string());
+        query_builder.push(order.sort_by.to_string().replace("ast_", ""));
         query_builder.push(direction);
-        query_builder.push(", ast_pubkey ");
+        query_builder.push(", pubkey ");
         query_builder.push(direction);
+
         // Add LIMIT clause
         query_builder.push(" LIMIT ");
         query_builder.push_bind(limit as i64);
@@ -168,20 +208,6 @@ fn add_filter_clause<'a>(
         query_builder.push(" INNER JOIN assets_authorities ON assets_v3.ast_authority_fk = assets_authorities.auth_pubkey ");
         group_clause_required = true;
     }
-    if let Some(ref token_type) = filter.token_type {
-        if token_type == &TokenType::All && filter.owner_address.is_some() {
-            query_builder.push(
-                " LEFT JOIN fungible_tokens ON assets_v3.ast_pubkey = fungible_tokens.fbt_asset ",
-            );
-            group_clause_required = true;
-        }
-        if token_type == &TokenType::Fungible && filter.owner_address.is_some() {
-            query_builder.push(
-                " INNER JOIN fungible_tokens ON assets_v3.ast_pubkey = fungible_tokens.fbt_asset ",
-            );
-            group_clause_required = true;
-        }
-    }
 
     // todo: if we implement the additional params like negata and all/any switch, the true part and the AND prefix should be refactored
     query_builder.push(" WHERE TRUE ");
@@ -259,14 +285,7 @@ fn add_filter_clause<'a>(
                 TokenType::All => {
                     query_builder.push(" AND (assets_v3.ast_owner = ");
                     query_builder.push_bind(owner_address);
-                    query_builder.push(" OR (fungible_tokens.fbt_owner = ");
-                    query_builder.push_bind(owner_address);
-                    if !options.show_zero_balance {
-                        query_builder.push(" AND fungible_tokens.fbt_balance > ");
-                        query_builder.push_bind(0i64);
-                    }
-                    query_builder.push(" ) ");
-                    query_builder.push(" ) ");
+                    query_builder.push(")");
                 },
             }
         } else {

postgre-client/tests/asset_filter_client_test.rs

@@ -1,7 +1,10 @@
 #[cfg(feature = "integration_tests")]
 #[cfg(test)]
 mod tests {
-    use entities::{api_req_params::GetByMethodsOptions, enums::AssetType};
+    use entities::{
+        api_req_params::GetByMethodsOptions,
+        enums::{AssetType, TokenType},
+    };
     use postgre_client::{
         model::*,
         storage_traits::{AssetIndexStorage, AssetPubkeyFilteredFetcher},
@@ -314,4 +317,70 @@ mod tests {
         assert_eq!(res.len(), 101);
         env.teardown().await;
     }
+
+    #[tokio::test]
+    #[tracing_test::traced_test]
+    async fn test_search_for_fungible_asset() {
+        let cli = Cli::default();
+        let env = TestEnvironment::new(&cli).await;
+        let storage = &env.client;
+
+        // Generate nft asset
+        let asset_indexes = generate_asset_index_records(1);
+        let asset_index = &asset_indexes[0];
+        let common_owner = asset_index.owner;
+
+        // Generate fungible asset and make it part of the nft asset
+        let mut fungible_asset_indexes = generate_fungible_asset_index_records(1);
+        let fungible_asset_index = &mut fungible_asset_indexes[0];
+        fungible_asset_index.pubkey = asset_index.pubkey;
+        fungible_asset_index.owner = common_owner;
+
+        let last_known_key = generate_random_vec(8 + 8 + 32);
+
+        // Insert assets and last key
+        storage.update_nft_asset_indexes_batch(asset_indexes.as_slice()).await.unwrap();
+        storage.update_last_synced_key(&last_known_key, AssetType::NonFungible).await.unwrap();
+
+        // Insert fungible assets and last key
+        storage
+            .update_fungible_asset_indexes_batch(fungible_asset_indexes.as_slice())
+            .await
+            .unwrap();
+        storage.update_last_synced_key(&last_known_key, AssetType::Fungible).await.unwrap();
+
+        let cnt = env.count_rows_in_assets().await.unwrap();
+        assert_eq!(cnt, 1);
+        let cnt = env.count_rows_in_fungible_tokens().await.unwrap();
+        assert_eq!(cnt, 1);
+
+        let order = AssetSorting {
+            sort_by: AssetSortBy::SlotCreated,
+            sort_direction: AssetSortDirection::Asc,
+        };
+
+        let res = storage
+            .get_asset_pubkeys_filtered(
+                &SearchAssetsFilter {
+                    owner_address: Some(common_owner.unwrap().to_bytes().to_vec()),
+                    token_type: Some(TokenType::All),
+                    ..Default::default()
+                },
+                &order,
+                1,
+                None,
+                None,
+                None,
+                &GetByMethodsOptions {
+                    show_zero_balance: true,
+                    show_unverified_collections: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        assert_eq!(res.len(), 1);
+
+        env.teardown().await;
+    }
 }

postgre-client/tests/asset_index_client_test.rs

@@ -8,6 +8,7 @@ mod tests {
     use postgre_client::storage_traits::AssetIndexStorage;
     use rand::Rng;
     use setup::pg::*;
+    use solana_sdk::pubkey::Pubkey;
     use testcontainers::clients::Cli;
 
     #[tokio::test]
@@ -86,12 +87,12 @@ mod tests {
         // every asset_index will have 3 creators
         for asset_index in asset_indexes.iter_mut() {
             asset_index.creators.push(Creator {
-                creator: generate_random_pubkey(),
+                creator: Pubkey::new_unique(),
                 creator_verified: rand::thread_rng().gen_bool(0.5),
                 creator_share: 30,
             });
             asset_index.creators.push(Creator {
-                creator: generate_random_pubkey(),
+                creator: Pubkey::new_unique(),
                 creator_verified: rand::thread_rng().gen_bool(0.5),
                 creator_share: 30,
             });
@@ -119,7 +120,7 @@ mod tests {
             .map(|asset_index| {
                 let mut ai = asset_index.clone();
                 ai.creators.pop();
-                ai.creators[1].creator = generate_random_pubkey();
+                ai.creators[1].creator = Pubkey::new_unique();
                 ai.creators[0].creator_verified = !ai.creators[0].creator_verified;
                 ai
             })

rocks-db/src/column.rs

@@ -236,10 +236,8 @@ where
             )
             .into_iter()
             .map(|res| {
-                res.map_err(StorageError::from).and_then(|opt| {
-                    opt.map(|pinned| C::decode(pinned.as_ref()).map_err(StorageError::from))
-                        .transpose()
-                })
+                res.map_err(StorageError::from)
+                    .and_then(|opt| opt.map(|pinned| C::decode(pinned.as_ref())).transpose())
             })
             .collect()
     }

rocks-db/src/columns/asset.rs

@@ -1958,7 +1958,7 @@ pub fn merge_complete_details_fb_simple_raw<'a>(
             }
             // Merge authority
             if let Some(new_authority) = new_val.authority() {
-                if authority.map_or(true, |current_authority| {
+                if authority.is_none_or(|current_authority| {
                     new_authority.compare(&current_authority) == Ordering::Greater
                 }) {
                     authority = Some(new_authority);

rocks-db/src/sequence_consistent.rs

@@ -30,7 +30,7 @@ impl SequenceConsistentManager for Storage {
         let result = if gap_found {
             self.trees_gaps.put_async(tree, TreesGaps {}).await
         } else {
-            self.trees_gaps.delete(tree).map_err(Into::into)
+            self.trees_gaps.delete(tree)
         };
         if let Err(e) = result {
             error!("{} tree gap: {}", if gap_found { "Put" } else { "Delete" }, e);

tests/setup/src/pg.rs

@@ -2,7 +2,7 @@ use std::{collections::BTreeMap, path::PathBuf, str::FromStr, sync::Arc};
 
 use entities::{
     enums::*,
-    models::{AssetIndex, Creator, UrlWithStatus},
+    models::{AssetIndex, Creator, FungibleAssetIndex, UrlWithStatus},
 };
 use metrics_utils::red::RequestErrorDurationMetrics;
 use postgre_client::{storage_traits::AssetIndexStorage, PgClient};
@@ -187,24 +187,20 @@ pub fn generate_random_vec(n: usize) -> Vec<u8> {
     random_vector
 }
 
-pub fn generate_random_pubkey() -> Pubkey {
-    Pubkey::new_unique()
-}
-
 pub fn generate_asset_index_records(n: usize) -> Vec<AssetIndex> {
     let mut asset_indexes = Vec::new();
     for i in 0..n {
         let asset_index = AssetIndex {
-            pubkey: generate_random_pubkey(),
+            pubkey: Pubkey::new_unique(),
             specification_version: SpecificationVersions::V1,
             specification_asset_class: SpecificationAssetClass::Nft,
             royalty_target_type: RoyaltyTargetType::Creators,
             royalty_amount: 1,
             slot_created: (n - i) as i64,
-            owner: Some(generate_random_pubkey()),
-            delegate: Some(generate_random_pubkey()),
-            authority: Some(generate_random_pubkey()),
-            collection: Some(generate_random_pubkey()),
+            owner: Some(Pubkey::new_unique()),
+            delegate: Some(Pubkey::new_unique()),
+            authority: Some(Pubkey::new_unique()),
+            collection: Some(Pubkey::new_unique()),
             is_collection_verified: Some(rand::thread_rng().gen_bool(0.5)),
             is_burnt: false,
             is_compressible: false,
@@ -218,7 +214,7 @@ pub fn generate_asset_index_records(n: usize) -> Vec<AssetIndex> {
             update_authority: None,
             slot_updated: (n + 10 + i) as i64,
             creators: vec![Creator {
-                creator: generate_random_pubkey(),
+                creator: Pubkey::new_unique(),
                 creator_verified: rand::thread_rng().gen_bool(0.5),
                 creator_share: 100,
             }],
@@ -230,3 +226,18 @@ pub fn generate_asset_index_records(n: usize) -> Vec<AssetIndex> {
     }
     asset_indexes
 }
+
+pub fn generate_fungible_asset_index_records(n: usize) -> Vec<FungibleAssetIndex> {
+    let mut asset_indexes = Vec::new();
+    for i in 0..n {
+        let asset_index = FungibleAssetIndex {
+            pubkey: Pubkey::new_unique(),
+            owner: Some(Pubkey::new_unique()),
+            slot_updated: (n + 10 + i) as i64,
+            fungible_asset_mint: None,
+            fungible_asset_balance: Some(1),
+        };
+        asset_indexes.push(asset_index);
+    }
+    asset_indexes
+}