Rewrite query for token transfers on address to eliminate "or" (#9576)

* Rewrite query for token transfers on address to eliminate "or"

* Add token_transfers [:to_address_hash, :block_number] index

* Review processing

* Review processing #2

* Update apps/explorer/lib/explorer/chain/token_transfer.ex

Co-authored-by: Kirill Fedoseev <kirill@blockscout.com>

---------

Co-authored-by: Kirill Fedoseev <kirill@blockscout.com>

CHANGELOG.md

@@ -23,6 +23,7 @@
 - [#9597](https://github.com/blockscout/blockscout/pull/9597) - Update token transfers block_consensus by block_number
 - [#9596](https://github.com/blockscout/blockscout/pull/9596) - Fix logging
 - [#9585](https://github.com/blockscout/blockscout/pull/9585) - Fix Geth block internal transactions fetching
+- [#9576](https://github.com/blockscout/blockscout/pull/9576) - Rewrite query for token transfers on address to eliminate "or"
 - [#9572](https://github.com/blockscout/blockscout/pull/9572) - Fix Shibarium L1 fetcher
 - [#9563](https://github.com/blockscout/blockscout/pull/9563) - Fix timestamp handler for unfinalized zkEVM batches
 - [#9560](https://github.com/blockscout/blockscout/pull/9560) - Fix fetch pending transaction for hyperledger besu client

apps/block_scout_web/lib/block_scout_web/controllers/api/v2/address_controller.ex

@@ -48,7 +48,8 @@ defmodule BlockScoutWeb.API.V2.AddressController do
       :to_address => :optional,
       :from_address => :optional,
       :block => :optional,
-      :transaction => :optional
+      :transaction => :optional,
+      :token => :optional
     },
     api?: true
   ]

apps/explorer/lib/explorer/chain.ex

@@ -307,9 +307,8 @@ defmodule Explorer.Chain do
     necessity_by_association = Keyword.get(options, :necessity_by_association)
 
     direction
-    |> TokenTransfer.token_transfers_by_address_hash(address_hash, filters)
+    |> TokenTransfer.token_transfers_by_address_hash(address_hash, filters, paging_options)
     |> join_associations(necessity_by_association)
-    |> TokenTransfer.handle_paging_options(paging_options)
     |> select_repo(options).all()
   end
 

apps/explorer/lib/explorer/chain/token_transfer.ex

@@ -311,13 +311,40 @@ defmodule Explorer.Chain.TokenTransfer do
     |> order_by([tt], desc: tt.block_number, desc: tt.log_index)
   end
 
-  def token_transfers_by_address_hash(direction, address_hash, token_types) do
-    only_consensus_transfers_query()
-    |> filter_by_direction(direction, address_hash)
-    |> order_by([tt], desc: tt.block_number, desc: tt.log_index)
-    |> join(:inner, [tt], token in assoc(tt, :token), as: :token)
-    |> preload([token: token], [{:token, token}])
-    |> filter_by_type(token_types)
+  def token_transfers_by_address_hash(direction, address_hash, token_types, paging_options) do
+    if direction == :to || direction == :from do
+      only_consensus_transfers_query()
+      |> filter_by_direction(direction, address_hash)
+      |> order_by([tt], desc: tt.block_number, desc: tt.log_index)
+      |> join(:inner, [tt], token in assoc(tt, :token), as: :token)
+      |> preload([token: token], [{:token, token}])
+      |> filter_by_type(token_types)
+      |> handle_paging_options(paging_options)
+    else
+      to_address_hash_query =
+        only_consensus_transfers_query()
+        |> join(:inner, [tt], token in assoc(tt, :token), as: :token)
+        |> filter_by_direction(:to, address_hash)
+        |> filter_by_type(token_types)
+        |> order_by([tt], desc: tt.block_number, desc: tt.log_index)
+        |> handle_paging_options(paging_options)
+        |> Chain.wrapped_union_subquery()
+
+      from_address_hash_query =
+        only_consensus_transfers_query()
+        |> join(:inner, [tt], token in assoc(tt, :token), as: :token)
+        |> filter_by_direction(:from, address_hash)
+        |> filter_by_type(token_types)
+        |> order_by([tt], desc: tt.block_number, desc: tt.log_index)
+        |> handle_paging_options(paging_options)
+        |> Chain.wrapped_union_subquery()
+
+      to_address_hash_query
+      |> union(^from_address_hash_query)
+      |> Chain.wrapped_union_subquery()
+      |> order_by([tt], desc: tt.block_number, desc: tt.log_index)
+      |> limit(^paging_options.page_size)
+    end
   end
 
   def filter_by_direction(query, :to, address_hash) do
@@ -330,11 +357,6 @@ defmodule Explorer.Chain.TokenTransfer do
     |> where([tt], tt.from_address_hash == ^address_hash)
   end
 
-  def filter_by_direction(query, _, address_hash) do
-    query
-    |> where([tt], tt.from_address_hash == ^address_hash or tt.to_address_hash == ^address_hash)
-  end
-
   def filter_by_type(query, []), do: query
 
   def filter_by_type(query, token_types) when is_list(token_types) do

apps/explorer/priv/repo/migrations/20240308123508_token_transfers_add_from_address_hash_block_number_index.exs

@@ -0,0 +1,9 @@
+defmodule Explorer.Repo.Migrations.TokenTransfersAddFromAddressHashBlockNumberIndex do
+  use Ecto.Migration
+  @disable_ddl_transaction true
+  @disable_migration_lock true
+
+  def change do
+    create_if_not_exists(index(:token_transfers, [:from_address_hash, :block_number], concurrently: true))
+  end
+end

apps/explorer/priv/repo/migrations/20240313195728_token_transfers_add_to_address_hash_block_number_index.exs

@@ -0,0 +1,9 @@
+defmodule Explorer.Repo.Migrations.TokenTransfersAddToAddressHashBlockNumberIndex do
+  use Ecto.Migration
+  @disable_ddl_transaction true
+  @disable_migration_lock true
+
+  def change do
+    create_if_not_exists(index(:token_transfers, [:to_address_hash, :block_number], concurrently: true))
+  end
+end

apps/indexer/lib/indexer/transform/token_transfers.ex

@@ -178,13 +178,16 @@ defmodule Indexer.Transform.TokenTransfers do
        when not is_nil(second_topic) and not is_nil(third_topic) do
     [amount] = Helper.decode_data(log.data, [{:uint, 256}])
 
+    from_address_hash = truncate_address_hash(log.second_topic)
+    to_address_hash = truncate_address_hash(log.third_topic)
+
     token_transfer = %{
       amount: Decimal.new(amount || 0),
       block_number: log.block_number,
       block_hash: log.block_hash,
       log_index: log.index,
-      from_address_hash: truncate_address_hash(log.second_topic),
-      to_address_hash: truncate_address_hash(log.third_topic),
+      from_address_hash: from_address_hash,
+      to_address_hash: to_address_hash,
       token_contract_address_hash: log.address_hash,
       transaction_hash: log.transaction_hash,
       token_ids: nil,
@@ -237,12 +240,15 @@ defmodule Indexer.Transform.TokenTransfers do
        when not is_nil(second_topic) and not is_nil(third_topic) and not is_nil(fourth_topic) do
     [token_id] = Helper.decode_data(fourth_topic, [{:uint, 256}])
 
+    from_address_hash = truncate_address_hash(log.second_topic)
+    to_address_hash = truncate_address_hash(log.third_topic)
+
     token_transfer = %{
       block_number: log.block_number,
       log_index: log.index,
       block_hash: log.block_hash,
-      from_address_hash: truncate_address_hash(log.second_topic),
-      to_address_hash: truncate_address_hash(log.third_topic),
+      from_address_hash: from_address_hash,
+      to_address_hash: to_address_hash,
       token_contract_address_hash: log.address_hash,
       token_ids: [token_id || 0],
       transaction_hash: log.transaction_hash,
@@ -302,12 +308,15 @@ defmodule Indexer.Transform.TokenTransfers do
     if is_nil(token_ids) or token_ids == [] or is_nil(values) or values == [] do
       nil
     else
+      from_address_hash = truncate_address_hash(third_topic)
+      to_address_hash = truncate_address_hash(fourth_topic)
+
       token_transfer = %{
         block_number: log.block_number,
         block_hash: log.block_hash,
         log_index: log.index,
-        from_address_hash: truncate_address_hash(third_topic),
-        to_address_hash: truncate_address_hash(fourth_topic),
+        from_address_hash: from_address_hash,
+        to_address_hash: to_address_hash,
         token_contract_address_hash: log.address_hash,
         transaction_hash: log.transaction_hash,
         token_type: "ERC-1155",
@@ -327,13 +336,16 @@ defmodule Indexer.Transform.TokenTransfers do
   def parse_erc1155_params(%{third_topic: third_topic, fourth_topic: fourth_topic, data: data} = log) do
     [token_id, value] = Helper.decode_data(data, [{:uint, 256}, {:uint, 256}])
 
+    from_address_hash = truncate_address_hash(third_topic)
+    to_address_hash = truncate_address_hash(fourth_topic)
+
     token_transfer = %{
       amount: value,
       block_number: log.block_number,
       block_hash: log.block_hash,
       log_index: log.index,
-      from_address_hash: truncate_address_hash(third_topic),
-      to_address_hash: truncate_address_hash(fourth_topic),
+      from_address_hash: from_address_hash,
+      to_address_hash: to_address_hash,
       token_contract_address_hash: log.address_hash,
       transaction_hash: log.transaction_hash,
       token_type: "ERC-1155",