Reorganize queries and indexes for internal_transactions table

CHANGELOG.md

@@ -20,6 +20,8 @@
 - [#2899](https://github.com/poanetwork/blockscout/pull/2899) - fix empty buffered task
 - [#2887](https://github.com/poanetwork/blockscout/pull/2887) - increase chart loading speed
 
+- [2910](https://github.com/poanetwork/blockscout/pull/2910) - Reorganize queries and indexes for internal_transactions table
+
 ### Chore
 - [#2896](https://github.com/poanetwork/blockscout/pull/2896) - Disable Parity websockets tests
 

apps/block_scout_web/test/block_scout_web/controllers/api/rpc/address_controller_test.exs

@@ -1626,6 +1626,7 @@ defmodule BlockScoutWeb.API.RPC.AddressControllerTest do
           transaction: transaction,
           index: 0,
           from_address: address,
+          block_number: block.number,
           block_hash: transaction.block_hash,
           block_index: 0
         )
@@ -1681,6 +1682,7 @@ defmodule BlockScoutWeb.API.RPC.AddressControllerTest do
         index: 0,
         type: :reward,
         error: "some error",
+        block_number: transaction.block_number,
         block_hash: transaction.block_hash,
         block_index: 0
       ]
@@ -1718,6 +1720,7 @@ defmodule BlockScoutWeb.API.RPC.AddressControllerTest do
           from_address: address,
           transaction: transaction,
           index: index,
+          block_number: transaction.block_number,
           block_hash: transaction.block_hash,
           block_index: index
         }

apps/block_scout_web/test/block_scout_web/controllers/transaction_internal_transaction_controller_test.exs

@@ -68,7 +68,8 @@ defmodule BlockScoutWeb.TransactionInternalTransactionControllerTest do
 
       assert json_response(conn, 200)
 
-      assert Enum.count(items) == 2
+      # excluding of internal transactions with type=call and index=0
+      assert Enum.count(items) == 1
     end
 
     test "includes USD exchange rate value for address in assigns", %{conn: conn} do

apps/explorer/lib/explorer/chain.ex

@@ -15,6 +15,7 @@ defmodule Explorer.Chain do
       preload: 2,
       select: 2,
       subquery: 1,
+      union: 2,
       union_all: 2,
       where: 2,
       where: 3
@@ -194,9 +195,66 @@ defmodule Explorer.Chain do
     direction = Keyword.get(options, :direction)
     paging_options = Keyword.get(options, :paging_options, @default_paging_options)
 
-    InternalTransaction
-    |> InternalTransaction.where_nonpending_block()
-    |> InternalTransaction.where_address_fields_match(hash, direction)
+    if direction == nil do
+      query_to_address_hash_wrapped =
+        InternalTransaction
+        |> InternalTransaction.where_address_fields_match(hash, :to_address_hash)
+        |> common_where_limit_order(paging_options)
+        |> wrapped_union_subquery()
+
+      query_from_address_hash_wrapped =
+        InternalTransaction
+        |> InternalTransaction.where_address_fields_match(hash, :from_address_hash)
+        |> common_where_limit_order(paging_options)
+        |> wrapped_union_subquery()
+
+      query_created_contract_address_hash_wrapped =
+        InternalTransaction
+        |> InternalTransaction.where_address_fields_match(hash, :created_contract_address_hash)
+        |> common_where_limit_order(paging_options)
+        |> wrapped_union_subquery()
+
+      full_query =
+        query_to_address_hash_wrapped
+        |> union(^query_from_address_hash_wrapped)
+        |> union(^query_created_contract_address_hash_wrapped)
+
+      full_wrapped_query =
+        from(
+          q in subquery(full_query),
+          select: q
+        )
+
+      full_wrapped_query
+      |> order_by(
+        [q],
+        desc: q.block_number,
+        desc: q.transaction_index,
+        desc: q.index
+      )
+      |> preload(transaction: :block)
+      |> join_associations(necessity_by_association)
+      |> Repo.all()
+    else
+      InternalTransaction
+      |> InternalTransaction.where_nonpending_block()
+      |> InternalTransaction.where_address_fields_match(hash, direction)
+      |> common_where_limit_order(paging_options)
+      |> preload(transaction: :block)
+      |> join_associations(necessity_by_association)
+      |> Repo.all()
+    end
+  end
+
+  def wrapped_union_subquery(query) do
+    from(
+      q in subquery(query),
+      select: q
+    )
+  end
+
+  defp common_where_limit_order(query, paging_options) do
+    query
     |> InternalTransaction.where_is_different_from_parent_transaction()
     |> InternalTransaction.where_block_number_is_not_null()
     |> page_internal_transaction(paging_options)
@@ -207,9 +265,6 @@ defmodule Explorer.Chain do
       desc: it.transaction_index,
       desc: it.index
     )
-    |> preload(transaction: :block)
-    |> join_associations(necessity_by_association)
-    |> Repo.all()
   end
 
   @doc """
@@ -2449,6 +2504,7 @@ defmodule Explorer.Chain do
     |> for_parent_transaction(hash)
     |> join_associations(necessity_by_association)
     |> where_transaction_has_multiple_internal_transactions()
+    |> InternalTransaction.where_is_different_from_parent_transaction()
     |> InternalTransaction.where_nonpending_block()
     |> page_internal_transaction(paging_options)
     |> limit(^paging_options.page_size)

apps/explorer/lib/explorer/chain/internal_transaction.ex

@@ -555,6 +555,18 @@ defmodule Explorer.Chain.InternalTransaction do
     )
   end
 
+  def where_address_fields_match(query, address_hash, :to_address_hash) do
+    where(query, [it], it.to_address_hash == ^address_hash)
+  end
+
+  def where_address_fields_match(query, address_hash, :from_address_hash) do
+    where(query, [it], it.from_address_hash == ^address_hash)
+  end
+
+  def where_address_fields_match(query, address_hash, :created_contract_address_hash) do
+    where(query, [it], it.created_contract_address_hash == ^address_hash)
+  end
+
   def where_is_different_from_parent_transaction(query) do
     where(
       query,

apps/explorer/lib/explorer/etherscan.ex

@@ -3,7 +3,7 @@ defmodule Explorer.Etherscan do
   The etherscan context.
   """
 
-  import Ecto.Query, only: [from: 2, where: 3, or_where: 3]
+  import Ecto.Query, only: [from: 2, where: 3, or_where: 3, union: 2]
 
   alias Explorer.Etherscan.Logs
   alias Explorer.{Chain, Repo}
@@ -97,6 +97,7 @@ defmodule Explorer.Etherscan do
 
     query
     |> Chain.where_transaction_has_multiple_internal_transactions()
+    |> InternalTransaction.where_is_different_from_parent_transaction()
     |> InternalTransaction.where_nonpending_block()
     |> Repo.all()
   end
@@ -121,28 +122,92 @@ defmodule Explorer.Etherscan do
       ) do
     options = Map.merge(@default_options, raw_options)
 
-    query =
+    direction =
+      case options do
+        %{filter_by: "to"} -> :to
+        %{filter_by: "from"} -> :from
+        _ -> nil
+      end
+
+    consensus_blocks =
       from(
-        it in InternalTransaction,
-        inner_join: t in assoc(it, :transaction),
-        inner_join: b in assoc(t, :block),
-        order_by: [{^options.order_by_direction, t.block_number}],
-        limit: ^options.page_size,
-        offset: ^offset(options),
-        select:
-          merge(map(it, ^@internal_transaction_fields), %{
-            block_timestamp: b.timestamp,
-            block_number: b.number
-          })
+        b in Block,
+        where: b.consensus == true
       )
 
-    query
-    |> Chain.where_transaction_has_multiple_internal_transactions()
-    |> where_address_match(address_hash, options)
-    |> where_start_block_match(options)
-    |> where_end_block_match(options)
-    |> InternalTransaction.where_nonpending_block()
-    |> Repo.all()
+    if direction == nil do
+      query =
+        from(
+          it in InternalTransaction,
+          inner_join: b in subquery(consensus_blocks),
+          on: it.block_number == b.number,
+          order_by: [
+            {^options.order_by_direction, it.block_number},
+            {:desc, it.transaction_index},
+            {:desc, it.index}
+          ],
+          limit: ^options.page_size,
+          offset: ^offset(options),
+          select:
+            merge(map(it, ^@internal_transaction_fields), %{
+              block_timestamp: b.timestamp,
+              block_number: b.number
+            })
+        )
+
+      query_to_address_hash_wrapped =
+        query
+        |> InternalTransaction.where_address_fields_match(address_hash, :to_address_hash)
+        |> InternalTransaction.where_is_different_from_parent_transaction()
+        |> where_start_block_match(options)
+        |> where_end_block_match(options)
+        |> Chain.wrapped_union_subquery()
+
+      query_from_address_hash_wrapped =
+        query
+        |> InternalTransaction.where_address_fields_match(address_hash, :from_address_hash)
+        |> InternalTransaction.where_is_different_from_parent_transaction()
+        |> where_start_block_match(options)
+        |> where_end_block_match(options)
+        |> Chain.wrapped_union_subquery()
+
+      query_created_contract_address_hash_wrapped =
+        query
+        |> InternalTransaction.where_address_fields_match(address_hash, :created_contract_address_hash)
+        |> InternalTransaction.where_is_different_from_parent_transaction()
+        |> where_start_block_match(options)
+        |> where_end_block_match(options)
+        |> Chain.wrapped_union_subquery()
+
+      query_to_address_hash_wrapped
+      |> union(^query_from_address_hash_wrapped)
+      |> union(^query_created_contract_address_hash_wrapped)
+      |> Repo.all()
+    else
+      query =
+        from(
+          it in InternalTransaction,
+          inner_join: t in assoc(it, :transaction),
+          inner_join: b in assoc(t, :block),
+          order_by: [{^options.order_by_direction, t.block_number}],
+          limit: ^options.page_size,
+          offset: ^offset(options),
+          select:
+            merge(map(it, ^@internal_transaction_fields), %{
+              block_timestamp: b.timestamp,
+              block_number: b.number
+            })
+        )
+
+      query
+      |> Chain.where_transaction_has_multiple_internal_transactions()
+      |> InternalTransaction.where_address_fields_match(address_hash, direction)
+      |> InternalTransaction.where_is_different_from_parent_transaction()
+      |> where_start_block_match(options)
+      |> where_end_block_match(options)
+      |> InternalTransaction.where_nonpending_block()
+      |> Repo.all()
+    end
   end
 
   @doc """

apps/explorer/lib/explorer/etherscan/logs.ex

@@ -5,10 +5,10 @@ defmodule Explorer.Etherscan.Logs do
 
   """
 
-  import Ecto.Query, only: [from: 2, where: 3, subquery: 1, order_by: 3]
+  import Ecto.Query, only: [from: 2, where: 3, subquery: 1, order_by: 3, union: 2]
 
+  alias Explorer.{Chain, Repo}
   alias Explorer.Chain.{Block, InternalTransaction, Log, Transaction}
-  alias Explorer.Repo
 
   @base_filter %{
     from_block: nil,
@@ -78,25 +78,25 @@ defmodule Explorer.Etherscan.Logs do
 
     logs_query = where_topic_match(Log, prepared_filter)
 
+    query_to_address_hash_wrapped =
+      logs_query
+      |> internal_transaction_query(:to_address_hash, prepared_filter, address_hash)
+      |> Chain.wrapped_union_subquery()
+
+    query_from_address_hash_wrapped =
+      logs_query
+      |> internal_transaction_query(:from_address_hash, prepared_filter, address_hash)
+      |> Chain.wrapped_union_subquery()
+
+    query_created_contract_address_hash_wrapped =
+      logs_query
+      |> internal_transaction_query(:created_contract_address_hash, prepared_filter, address_hash)
+      |> Chain.wrapped_union_subquery()
+
     internal_transaction_log_query =
-      from(internal_transaction in InternalTransaction.where_nonpending_block(),
-        join: transaction in assoc(internal_transaction, :transaction),
-        join: log in ^logs_query,
-        on: log.transaction_hash == internal_transaction.transaction_hash,
-        where: internal_transaction.block_number >= ^prepared_filter.from_block,
-        where: internal_transaction.block_number <= ^prepared_filter.to_block,
-        where:
-          internal_transaction.to_address_hash == ^address_hash or
-            internal_transaction.from_address_hash == ^address_hash or
-            internal_transaction.created_contract_address_hash == ^address_hash,
-        select:
-          merge(map(log, ^@log_fields), %{
-            gas_price: transaction.gas_price,
-            gas_used: transaction.gas_used,
-            transaction_index: transaction.index,
-            block_number: transaction.block_number
-          })
-      )
+      query_to_address_hash_wrapped
+      |> union(^query_from_address_hash_wrapped)
+      |> union(^query_created_contract_address_hash_wrapped)
 
     all_transaction_logs_query =
       from(transaction in Transaction,
@@ -256,4 +256,25 @@ defmodule Explorer.Etherscan.Logs do
           data.transaction_index >= ^transaction_index
     )
   end
+
+  defp internal_transaction_query(logs_query, direction, prepared_filter, address_hash) do
+    query =
+      from(internal_transaction in InternalTransaction.where_nonpending_block(),
+        join: transaction in assoc(internal_transaction, :transaction),
+        join: log in ^logs_query,
+        on: log.transaction_hash == internal_transaction.transaction_hash,
+        where: internal_transaction.block_number >= ^prepared_filter.from_block,
+        where: internal_transaction.block_number <= ^prepared_filter.to_block,
+        select:
+          merge(map(log, ^@log_fields), %{
+            gas_price: transaction.gas_price,
+            gas_used: transaction.gas_used,
+            transaction_index: transaction.index,
+            block_number: transaction.block_number
+          })
+      )
+
+    query
+    |> InternalTransaction.where_address_fields_match(address_hash, direction)
+  end
 end

apps/explorer/priv/repo/migrations/20191203112646_internal_transactions_add_to_address_hash_index.exs

@@ -0,0 +1,25 @@
+defmodule Explorer.Repo.Migrations.InternalTransactionsAddToAddressHashIndex do
+  use Ecto.Migration
+
+  def change do
+    execute(
+      "CREATE INDEX IF NOT EXISTS internal_transactions_from_address_hash_partial_index on public.internal_transactions(from_address_hash, block_number DESC, transaction_index DESC, index DESC) WHERE (((type = 'call') AND (index > 0)) OR (type != 'call'));"
+    )
+
+    execute(
+      "CREATE INDEX IF NOT EXISTS internal_transactions_to_address_hash_partial_index on public.internal_transactions(to_address_hash, block_number DESC, transaction_index DESC, index DESC) WHERE (((type = 'call') AND (index > 0)) OR (type != 'call'));"
+    )
+
+    execute(
+      "CREATE INDEX IF NOT EXISTS internal_transactions_created_contract_address_hash_partial_index on public.internal_transactions(created_contract_address_hash, block_number DESC, transaction_index DESC, index DESC) WHERE (((type = 'call') AND (index > 0)) OR (type != 'call'));"
+    )
+
+    drop_if_exists(
+      index(
+        :internal_transactions,
+        [:to_address_hash, :from_address_hash, :created_contract_address_hash, :type, :index],
+        name: "internal_transactions_to_address_hash_from_address_hash_created"
+      )
+    )
+  end
+end