[indexer-alt] add transaction indices pipelines (#19953)

## Description 

This PR adds schema and pipeline for various transaction indices
including
- tx_affected_addresses (replacing tx_senders and tx_recipients)
- tx_kinds
- tx_digests
- tx_calls_fun (collapsing into one)

## Test plan 

cargo run --                                             \          
--database-url
"postgres://postgres:postgrespw@localhost:5432/sui_indexer_alt" \
--remote-store-url https://checkpoints.mainnet.sui.io \
  --last-checkpoint 10000

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] Indexer: 
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

---------

Co-authored-by: Ashok Menon <[email protected]>

crates/sui-indexer-alt/Cargo.toml

@@ -23,6 +23,7 @@ diesel = { workspace = true, features = ["chrono"] }
 diesel-async = { workspace = true, features = ["bb8", "postgres", "async-connection-wrapper"] }
 diesel_migrations.workspace = true
 futures.workspace = true
+itertools.workspace = true
 prometheus.workspace = true
 reqwest.workspace = true
 serde.workspace = true

crates/sui-indexer-alt/migrations/2024-10-16-002409_tx_affected_objects/up.sql

@@ -1,4 +1,5 @@
-CREATE TABLE IF NOT EXISTS tx_affected_objects (
+CREATE TABLE IF NOT EXISTS tx_affected_objects
+(
     tx_sequence_number          BIGINT       NOT NULL,
     -- Object ID of the object touched by this transaction.
     affected                    BYTEA        NOT NULL,

crates/sui-indexer-alt/migrations/2024-10-21-003426_tx_indices/down.sql

@@ -0,0 +1,4 @@
+DROP TABLE IF EXISTS tx_affected_addresses;
+DROP TABLE IF EXISTS tx_calls_fun;
+DROP TABLE IF EXISTS tx_digests;
+DROP TABLE IF EXISTS tx_kinds;

crates/sui-indexer-alt/migrations/2024-10-21-003426_tx_indices/up.sql

@@ -0,0 +1,57 @@
+CREATE TABLE IF NOT EXISTS tx_affected_addresses
+(
+    affected                    BYTEA        NOT NULL,
+    tx_sequence_number          BIGINT       NOT NULL,
+    sender                      BYTEA        NOT NULL,
+    PRIMARY KEY (affected, tx_sequence_number)
+);
+
+CREATE INDEX IF NOT EXISTS tx_affected_addresses_tx_sequence_number
+ON tx_affected_addresses (tx_sequence_number);
+
+CREATE INDEX IF NOT EXISTS tx_affected_addresses_sender
+ON tx_affected_addresses (sender, affected, tx_sequence_number);
+
+CREATE TABLE IF NOT EXISTS tx_digests
+(
+    tx_sequence_number          BIGINT       PRIMARY KEY,
+    tx_digest                   BYTEA        NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS tx_kinds
+(
+    tx_kind                     SMALLINT     NOT NULL,
+    tx_sequence_number          BIGINT       NOT NULL,
+    PRIMARY KEY (tx_kind, tx_sequence_number)
+);
+
+CREATE INDEX IF NOT EXISTS tx_kinds_tx_sequence_number
+ON tx_kinds (tx_sequence_number);
+
+CREATE TABLE IF NOT EXISTS tx_calls
+(
+    package                     BYTEA        NOT NULL,
+    module                      TEXT         NOT NULL,
+    function                    TEXT         NOT NULL,
+    tx_sequence_number          BIGINT       NOT NULL,
+    sender                      BYTEA        NOT NULL,
+    PRIMARY KEY (package, module, function, tx_sequence_number)
+);
+
+CREATE INDEX IF NOT EXISTS tx_calls_tx_sequence_number
+ON tx_calls (tx_sequence_number);
+
+CREATE INDEX IF NOT EXISTS tx_calls_fun_sender
+ON tx_calls (sender, package, module, function, tx_sequence_number);
+
+CREATE INDEX IF NOT EXISTS tx_calls_mod
+ON tx_calls (package, module, tx_sequence_number);
+
+CREATE INDEX IF NOT EXISTS tx_calls_mod_sender
+ON tx_calls (sender, package, module, tx_sequence_number);
+
+CREATE INDEX IF NOT EXISTS tx_calls_pkg
+ON tx_calls (package, tx_sequence_number);
+
+CREATE INDEX IF NOT EXISTS tx_calls_pkg_sender
+ON tx_calls (sender, package, tx_sequence_number);

crates/sui-indexer-alt/src/handlers/mod.rs

@@ -9,7 +9,11 @@ pub mod kv_transactions;
 pub mod obj_versions;
 pub mod sum_coin_balances;
 pub mod sum_obj_types;
+pub mod tx_affected_addresses;
 pub mod tx_affected_objects;
 pub mod tx_balance_changes;
+pub mod tx_calls_fun;
+pub mod tx_digests;
+pub mod tx_kinds;
 pub mod wal_coin_balances;
 pub mod wal_obj_types;

crates/sui-indexer-alt/src/handlers/tx_affected_addresses.rs

@@ -0,0 +1,73 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::sync::Arc;
+
+use anyhow::Result;
+use diesel_async::RunQueryDsl;
+use itertools::Itertools;
+use sui_types::{full_checkpoint_content::CheckpointData, object::Owner};
+
+use crate::{
+    db, models::transactions::StoredTxAffectedAddress, pipeline::concurrent::Handler,
+    pipeline::Processor, schema::tx_affected_addresses,
+};
+
+pub struct TxAffectedAddress;
+
+impl Processor for TxAffectedAddress {
+    const NAME: &'static str = "tx_affected_addresses";
+
+    type Value = StoredTxAffectedAddress;
+
+    fn process(checkpoint: &Arc<CheckpointData>) -> Result<Vec<Self::Value>> {
+        let CheckpointData {
+            transactions,
+            checkpoint_summary,
+            ..
+        } = checkpoint.as_ref();
+
+        let mut values = Vec::new();
+        let first_tx = checkpoint_summary.network_total_transactions as usize - transactions.len();
+
+        for (i, tx) in transactions.iter().enumerate() {
+            let tx_sequence_number = (first_tx + i) as i64;
+            let sender = tx.transaction.sender_address();
+            let payer = tx.transaction.gas_owner();
+            let recipients = tx.effects.all_changed_objects().into_iter().filter_map(
+                |(_object_ref, owner, _write_kind)| match owner {
+                    Owner::AddressOwner(address) => Some(address),
+                    _ => None,
+                },
+            );
+
+            let affected_addresses: Vec<StoredTxAffectedAddress> = recipients
+                .chain(vec![sender, payer])
+                .unique()
+                .map(|a| StoredTxAffectedAddress {
+                    tx_sequence_number,
+                    affected: a.to_vec(),
+                    sender: sender.to_vec(),
+                })
+                .collect();
+            values.extend(affected_addresses);
+        }
+
+        Ok(values)
+    }
+}
+
+#[async_trait::async_trait]
+impl Handler for TxAffectedAddress {
+    const MIN_EAGER_ROWS: usize = 100;
+    const MAX_CHUNK_ROWS: usize = 1000;
+    const MAX_PENDING_ROWS: usize = 10000;
+
+    async fn commit(values: &[Self::Value], conn: &mut db::Connection<'_>) -> Result<usize> {
+        Ok(diesel::insert_into(tx_affected_addresses::table)
+            .values(values)
+            .on_conflict_do_nothing()
+            .execute(conn)
+            .await?)
+    }
+}

crates/sui-indexer-alt/src/handlers/tx_calls_fun.rs

@@ -0,0 +1,68 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::sync::Arc;
+
+use anyhow::{Ok, Result};
+use diesel_async::RunQueryDsl;
+use sui_types::full_checkpoint_content::CheckpointData;
+use sui_types::transaction::TransactionDataAPI;
+
+use crate::{
+    db, models::transactions::StoredTxCalls, pipeline::concurrent::Handler, pipeline::Processor,
+    schema::tx_calls,
+};
+
+pub struct TxCallsFun;
+
+impl Processor for TxCallsFun {
+    const NAME: &'static str = "tx_calls_fun";
+
+    type Value = StoredTxCalls;
+
+    fn process(checkpoint: &Arc<CheckpointData>) -> Result<Vec<Self::Value>> {
+        let CheckpointData {
+            transactions,
+            checkpoint_summary,
+            ..
+        } = checkpoint.as_ref();
+
+        let first_tx = checkpoint_summary.network_total_transactions as usize - transactions.len();
+
+        Ok(transactions
+            .iter()
+            .enumerate()
+            .flat_map(|(i, tx)| {
+                let tx_sequence_number = (first_tx + i) as i64;
+                let sender = tx.transaction.sender_address().to_vec();
+                let calls = tx.transaction.data().transaction_data().move_calls();
+
+                calls
+                    .iter()
+                    .map(|(package, module, function)| StoredTxCalls {
+                        tx_sequence_number,
+                        package: package.to_vec(),
+                        module: module.to_string(),
+                        function: function.to_string(),
+                        sender: sender.clone(),
+                    })
+                    .collect::<Vec<_>>()
+            })
+            .collect())
+    }
+}
+
+#[async_trait::async_trait]
+impl Handler for TxCallsFun {
+    const MIN_EAGER_ROWS: usize = 100;
+    const MAX_CHUNK_ROWS: usize = 1000;
+    const MAX_PENDING_ROWS: usize = 10000;
+
+    async fn commit(values: &[Self::Value], conn: &mut db::Connection<'_>) -> Result<usize> {
+        Ok(diesel::insert_into(tx_calls::table)
+            .values(values)
+            .on_conflict_do_nothing()
+            .execute(conn)
+            .await?)
+    }
+}

crates/sui-indexer-alt/src/handlers/tx_digests.rs

@@ -0,0 +1,55 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::sync::Arc;
+
+use anyhow::Result;
+use diesel_async::RunQueryDsl;
+use sui_types::full_checkpoint_content::CheckpointData;
+
+use crate::{
+    db, models::transactions::StoredTxDigest, pipeline::concurrent::Handler, pipeline::Processor,
+    schema::tx_digests,
+};
+
+pub struct TxDigests;
+
+impl Processor for TxDigests {
+    const NAME: &'static str = "tx_digests";
+
+    type Value = StoredTxDigest;
+
+    fn process(checkpoint: &Arc<CheckpointData>) -> Result<Vec<Self::Value>> {
+        let CheckpointData {
+            transactions,
+            checkpoint_summary,
+            ..
+        } = checkpoint.as_ref();
+
+        let first_tx = checkpoint_summary.network_total_transactions as usize - transactions.len();
+
+        Ok(transactions
+            .iter()
+            .enumerate()
+            .map(|(i, tx)| StoredTxDigest {
+                tx_sequence_number: (first_tx + i) as i64,
+                tx_digest: tx.transaction.digest().inner().to_vec(),
+            })
+            .collect())
+    }
+}
+
+#[async_trait::async_trait]
+impl Handler for TxDigests {
+    const MIN_EAGER_ROWS: usize = 100;
+    const MAX_CHUNK_ROWS: usize = 1000;
+    const MAX_PENDING_ROWS: usize = 10000;
+
+    async fn commit(values: &[Self::Value], conn: &mut db::Connection<'_>) -> Result<usize> {
+        Ok(diesel::insert_into(tx_digests::table)
+            .values(values)
+            .on_conflict_do_nothing()
+            .execute(conn)
+            .await?)
+    }
+}

crates/sui-indexer-alt/src/handlers/tx_kinds.rs

@@ -0,0 +1,65 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::sync::Arc;
+
+use anyhow::Result;
+use diesel_async::RunQueryDsl;
+use sui_types::full_checkpoint_content::CheckpointData;
+
+use crate::{
+    db,
+    models::transactions::{StoredKind, StoredTxKind},
+    pipeline::{concurrent::Handler, Processor},
+    schema::tx_kinds,
+};
+
+pub struct TxKinds;
+
+impl Processor for TxKinds {
+    const NAME: &'static str = "tx_kinds";
+
+    type Value = StoredTxKind;
+
+    fn process(checkpoint: &Arc<CheckpointData>) -> Result<Vec<Self::Value>> {
+        let CheckpointData {
+            transactions,
+            checkpoint_summary,
+            ..
+        } = checkpoint.as_ref();
+
+        let mut values = Vec::new();
+        let first_tx = checkpoint_summary.network_total_transactions as usize - transactions.len();
+
+        for (i, tx) in transactions.iter().enumerate() {
+            let tx_sequence_number = (first_tx + i) as i64;
+            let tx_kind = if tx.transaction.is_system_tx() {
+                StoredKind::SystemTransaction
+            } else {
+                StoredKind::ProgrammableTransaction
+            };
+
+            values.push(StoredTxKind {
+                tx_sequence_number,
+                tx_kind,
+            });
+        }
+
+        Ok(values)
+    }
+}
+
+#[async_trait::async_trait]
+impl Handler for TxKinds {
+    const MIN_EAGER_ROWS: usize = 100;
+    const MAX_CHUNK_ROWS: usize = 1000;
+    const MAX_PENDING_ROWS: usize = 10000;
+
+    async fn commit(values: &[Self::Value], conn: &mut db::Connection<'_>) -> Result<usize> {
+        Ok(diesel::insert_into(tx_kinds::table)
+            .values(values)
+            .on_conflict_do_nothing()
+            .execute(conn)
+            .await?)
+    }
+}

crates/sui-indexer-alt/src/main.rs

@@ -11,8 +11,9 @@ use sui_indexer_alt::{
         ev_emit_mod::EvEmitMod, ev_struct_inst::EvStructInst, kv_checkpoints::KvCheckpoints,
         kv_objects::KvObjects, kv_transactions::KvTransactions, obj_versions::ObjVersions,
         sum_coin_balances::SumCoinBalances, sum_obj_types::SumObjTypes,
-        tx_affected_objects::TxAffectedObjects, tx_balance_changes::TxBalanceChanges,
-        wal_coin_balances::WalCoinBalances, wal_obj_types::WalObjTypes,
+        tx_affected_addresses::TxAffectedAddress, tx_affected_objects::TxAffectedObjects,
+        tx_balance_changes::TxBalanceChanges, tx_calls_fun::TxCallsFun, tx_digests::TxDigests,
+        tx_kinds::TxKinds, wal_coin_balances::WalCoinBalances, wal_obj_types::WalObjTypes,
     },
     Indexer,
 };
@@ -42,8 +43,13 @@ async fn main() -> Result<()> {
             indexer.concurrent_pipeline::<KvObjects>().await?;
             indexer.concurrent_pipeline::<KvTransactions>().await?;
             indexer.concurrent_pipeline::<ObjVersions>().await?;
+            indexer.concurrent_pipeline::<TxAffectedAddress>().await?;
             indexer.concurrent_pipeline::<TxAffectedObjects>().await?;
             indexer.concurrent_pipeline::<TxBalanceChanges>().await?;
+            indexer.concurrent_pipeline::<TxCallsFun>().await?;
+            indexer.concurrent_pipeline::<TxDigests>().await?;
+            indexer.concurrent_pipeline::<TxKinds>().await?;
+            indexer.concurrent_pipeline::<TxKinds>().await?;
             indexer.concurrent_pipeline::<WalCoinBalances>().await?;
             indexer.concurrent_pipeline::<WalObjTypes>().await?;
             indexer.sequential_pipeline::<SumCoinBalances>(lag).await?;