feat(indexer): index tx_affected_objects (#19447)

## Description

Similar to #19355, introduce `tx_affected_objects` table -- a
combination of `tx_input_objects` and `tx_changed_objects` which will
both eventually be removed in favour of the new table.

## Test plan

```
sui$ cargo build -p sui-indexer
```

Tests based on reading this field will be included with the PR
introducing changes to GraphQL.

---

## 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): 
- [x] Indexer: Index the objects affected by a transaction (either
because they are an input object or are changed by the transaction).
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

crates/sui-indexer/migrations/pg/2024-09-19-121113_tx_affected_objects/down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS tx_affected_objects;

crates/sui-indexer/migrations/pg/2024-09-19-121113_tx_affected_objects/up.sql

@@ -0,0 +1,9 @@
+CREATE TABLE tx_affected_objects (
+    tx_sequence_number          BIGINT       NOT NULL,
+    affected                    BYTEA        NOT NULL,
+    sender                      BYTEA        NOT NULL,
+    PRIMARY KEY(affected, tx_sequence_number)
+);
+
+CREATE INDEX tx_affected_objects_tx_sequence_number_index ON tx_affected_objects (tx_sequence_number);
+CREATE INDEX tx_affected_objects_sender ON tx_affected_objects (sender, affected, tx_sequence_number);

crates/sui-indexer/src/models/tx_indices.rs

@@ -3,8 +3,8 @@
 
 use crate::{
     schema::{
-        tx_affected_addresses, tx_calls_fun, tx_calls_mod, tx_calls_pkg, tx_changed_objects,
-        tx_digests, tx_input_objects, tx_kinds, tx_recipients, tx_senders,
+        tx_affected_addresses, tx_affected_objects, tx_calls_fun, tx_calls_mod, tx_calls_pkg,
+        tx_changed_objects, tx_digests, tx_input_objects, tx_kinds, tx_recipients, tx_senders,
     },
     types::TxIndex,
 };
@@ -25,7 +25,15 @@ pub struct TxDigest {
 
 #[derive(Queryable, Insertable, Selectable, Debug, Clone, Default)]
 #[diesel(table_name = tx_affected_addresses)]
-pub struct StoredTxAffected {
+pub struct StoredTxAffectedAddresses {
+    pub tx_sequence_number: i64,
+    pub affected: Vec<u8>,
+    pub sender: Vec<u8>,
+}
+
+#[derive(Queryable, Insertable, Selectable, Debug, Clone, Default)]
+#[diesel(table_name = tx_affected_objects)]
+pub struct StoredTxAffectedObjects {
     pub tx_sequence_number: i64,
     pub affected: Vec<u8>,
     pub sender: Vec<u8>,
@@ -108,7 +116,8 @@ impl TxIndex {
     pub fn split(
         self: TxIndex,
     ) -> (
-        Vec<StoredTxAffected>,
+        Vec<StoredTxAffectedAddresses>,
+        Vec<StoredTxAffectedObjects>,
         Vec<StoredTxSenders>,
         Vec<StoredTxRecipients>,
         Vec<StoredTxInputObject>,
@@ -120,19 +129,32 @@ impl TxIndex {
         Vec<StoredTxKind>,
     ) {
         let tx_sequence_number = self.tx_sequence_number as i64;
+
         let tx_affected_addresses = self
             .recipients
             .iter()
             .chain(self.payers.iter())
             .chain(std::iter::once(&self.sender))
             .unique()
-            .map(|a| StoredTxAffected {
+            .map(|a| StoredTxAffectedAddresses {
                 tx_sequence_number,
                 affected: a.to_vec(),
                 sender: self.sender.to_vec(),
             })
             .collect();
 
+        let tx_affected_objects = self
+            .input_objects
+            .iter()
+            .chain(self.changed_objects.iter())
+            .unique()
+            .map(|o| StoredTxAffectedObjects {
+                tx_sequence_number,
+                affected: o.to_vec(),
+                sender: self.sender.to_vec(),
+            })
+            .collect();
+
         let tx_sender = StoredTxSenders {
             tx_sequence_number,
             sender: self.sender.to_vec(),
@@ -224,6 +246,7 @@ impl TxIndex {
 
         (
             tx_affected_addresses,
+            tx_affected_objects,
             vec![tx_sender],
             tx_recipients,
             tx_input_objects,

crates/sui-indexer/src/schema.rs

@@ -288,6 +288,14 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    tx_affected_objects (affected, tx_sequence_number) {
+        tx_sequence_number -> Int8,
+        affected -> Bytea,
+        sender -> Bytea,
+    }
+}
+
 diesel::table! {
     tx_calls_fun (package, module, func, tx_sequence_number) {
         tx_sequence_number -> Int8,
@@ -384,6 +392,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     pruner_cp_watermark,
     transactions,
     tx_affected_addresses,
+    tx_affected_objects,
     tx_calls_fun,
     tx_calls_mod,
     tx_calls_pkg,

crates/sui-indexer/src/store/pg_indexer_store.rs

@@ -50,8 +50,9 @@ use crate::schema::{
     event_senders, event_struct_instantiation, event_struct_module, event_struct_name,
     event_struct_package, events, feature_flags, full_objects_history, objects, objects_history,
     objects_snapshot, objects_version, packages, protocol_configs, pruner_cp_watermark,
-    transactions, tx_affected_addresses, tx_calls_fun, tx_calls_mod, tx_calls_pkg,
-    tx_changed_objects, tx_digests, tx_input_objects, tx_kinds, tx_recipients, tx_senders,
+    transactions, tx_affected_addresses, tx_affected_objects, tx_calls_fun, tx_calls_mod,
+    tx_calls_pkg, tx_changed_objects, tx_digests, tx_input_objects, tx_kinds, tx_recipients,
+    tx_senders,
 };
 use crate::store::transaction_with_retry;
 use crate::types::EventIndex;
@@ -1027,6 +1028,7 @@ impl PgIndexerStore {
         let len = indices.len();
         let (
             affected_addresses,
+            affected_objects,
             senders,
             recipients,
             input_objects,
@@ -1048,9 +1050,11 @@ impl PgIndexerStore {
                 Vec::new(),
                 Vec::new(),
                 Vec::new(),
+                Vec::new(),
             ),
             |(
                 mut tx_affected_addresses,
+                mut tx_affected_objects,
                 mut tx_senders,
                 mut tx_recipients,
                 mut tx_input_objects,
@@ -1063,17 +1067,19 @@ impl PgIndexerStore {
             ),
              index| {
                 tx_affected_addresses.extend(index.0);
-                tx_senders.extend(index.1);
-                tx_recipients.extend(index.2);
-                tx_input_objects.extend(index.3);
-                tx_changed_objects.extend(index.4);
-                tx_pkgs.extend(index.5);
-                tx_mods.extend(index.6);
-                tx_funs.extend(index.7);
-                tx_digests.extend(index.8);
-                tx_kinds.extend(index.9);
+                tx_affected_objects.extend(index.1);
+                tx_senders.extend(index.2);
+                tx_recipients.extend(index.3);
+                tx_input_objects.extend(index.4);
+                tx_changed_objects.extend(index.5);
+                tx_pkgs.extend(index.6);
+                tx_mods.extend(index.7);
+                tx_funs.extend(index.8);
+                tx_digests.extend(index.9);
+                tx_kinds.extend(index.10);
                 (
                     tx_affected_addresses,
+                    tx_affected_objects,
                     tx_senders,
                     tx_recipients,
                     tx_input_objects,
@@ -1095,6 +1101,12 @@ impl PgIndexerStore {
                     .execute(conn)
                     .await?;
 
+                diesel::insert_into(tx_affected_objects::table)
+                    .values(&affected_objects)
+                    .on_conflict_do_nothing()
+                    .execute(conn)
+                    .await?;
+
                 diesel::insert_into(tx_senders::table)
                     .values(&senders)
                     .on_conflict_do_nothing()