[indexer-alt] Add obj_info pipeline (MystenLabs#20436)

## Description 

Describe the changes or additions included in this PR.

## Test plan 

How did you test the new or updated feature?

---

## 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:

crates/sui-indexer-alt/migrations/2024-11-25-211949_obj_info/down.sql

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

crates/sui-indexer-alt/migrations/2024-11-25-211949_obj_info/up.sql

@@ -0,0 +1,67 @@
+-- A table that keeps track of all the updates to object type and owner information.
+-- In particular, whenever an object's presence or ownership changes, we insert a
+-- new row into this table. Each row should have a unique (object_id, cp_sequence_number)
+-- pair.
+-- When implementing consistency queries, we will use this table to find all
+-- object IDs that match the given filters bounded by the cursor checkpoint.
+-- These object IDs can then be used to look up the latest version of the objects
+-- bounded by the given checkpoint in the object_versions table.
+CREATE TABLE IF NOT EXISTS obj_info
+(
+    object_id                   BYTEA         NOT NULL,
+    cp_sequence_number          BIGINT        NOT NULL,
+    -- An enum describing the object's ownership model:
+    --
+    --   Immutable = 0,
+    --   Address-owned = 1,
+    --   Object-owned (dynamic field) = 2,
+    --   Shared = 3.
+    --
+    -- Note that there is a distinction between an object that is owned by
+    -- another object (kind 2), which relates to dynamic fields, and an object
+    -- that is owned by another object's address (kind 1), which relates to
+    -- transfer-to-object.
+    owner_kind                  SMALLINT,
+    -- The address for address-owned objects, and the parent object for
+    -- object-owned objects.
+    owner_id                    BYTEA,
+    -- The following fields relate to the object's type. These only apply to
+    -- Move Objects. For Move Packages they will all be NULL.
+    --
+    -- The type's package ID.
+    package                     BYTEA,
+    -- The type's module name.
+    module                      TEXT,
+    -- The type's name.
+    name                        TEXT,
+    -- The type's type parameters, as a BCS-encoded array of TypeTags.
+    instantiation               BYTEA,
+    PRIMARY KEY (object_id, cp_sequence_number)
+);
+
+CREATE INDEX IF NOT EXISTS obj_info_owner
+ON obj_info (owner_kind, owner_id, cp_sequence_number, object_id);
+
+CREATE INDEX IF NOT EXISTS obj_info_pkg
+ON obj_info (package, cp_sequence_number, object_id);
+
+CREATE INDEX IF NOT EXISTS obj_info_mod
+ON obj_info (package, module, cp_sequence_number, object_id);
+
+CREATE INDEX IF NOT EXISTS obj_info_name
+ON obj_info (package, module, name, cp_sequence_number, object_id);
+
+CREATE INDEX IF NOT EXISTS obj_info_inst
+ON obj_info (package, module, name, instantiation, cp_sequence_number, object_id);
+
+CREATE INDEX IF NOT EXISTS obj_info_owner_pkg
+ON obj_info (owner_kind, owner_id, package, cp_sequence_number, object_id);
+
+CREATE INDEX IF NOT EXISTS obj_info_owner_mod
+ON obj_info (owner_kind, owner_id, package, module, cp_sequence_number, object_id);
+
+CREATE INDEX IF NOT EXISTS obj_info_owner_name
+ON obj_info (owner_kind, owner_id, package, module, name, cp_sequence_number, object_id);
+
+CREATE INDEX IF NOT EXISTS obj_info_owner_inst
+ON obj_info (owner_kind, owner_id, package, module, name, instantiation, cp_sequence_number, object_id);

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

@@ -157,6 +157,7 @@ pub struct PipelineLayer {
     pub kv_objects: Option<ConcurrentLayer>,
     pub kv_protocol_configs: Option<ConcurrentLayer>,
     pub kv_transactions: Option<ConcurrentLayer>,
+    pub obj_info: Option<ConcurrentLayer>,
     pub obj_versions: Option<ConcurrentLayer>,
     pub tx_affected_addresses: Option<ConcurrentLayer>,
     pub tx_affected_objects: Option<ConcurrentLayer>,
@@ -293,6 +294,7 @@ impl PipelineLayer {
             kv_objects: Some(Default::default()),
             kv_protocol_configs: Some(Default::default()),
             kv_transactions: Some(Default::default()),
+            obj_info: Some(Default::default()),
             obj_versions: Some(Default::default()),
             tx_affected_addresses: Some(Default::default()),
             tx_affected_objects: Some(Default::default()),
@@ -430,6 +432,7 @@ impl Merge for PipelineLayer {
             kv_objects: self.kv_objects.merge(other.kv_objects),
             kv_protocol_configs: self.kv_protocol_configs.merge(other.kv_protocol_configs),
             kv_transactions: self.kv_transactions.merge(other.kv_transactions),
+            obj_info: self.obj_info.merge(other.obj_info),
             obj_versions: self.obj_versions.merge(other.obj_versions),
             tx_affected_addresses: self
                 .tx_affected_addresses

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

@@ -10,6 +10,7 @@ pub mod kv_feature_flags;
 pub mod kv_objects;
 pub mod kv_protocol_configs;
 pub mod kv_transactions;
+pub mod obj_info;
 pub mod obj_versions;
 pub mod sum_coin_balances;
 pub mod sum_displays;

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

@@ -0,0 +1,112 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::{collections::BTreeMap, sync::Arc};
+
+use anyhow::{anyhow, Result};
+use diesel_async::RunQueryDsl;
+use sui_types::{base_types::ObjectID, full_checkpoint_content::CheckpointData, object::Owner};
+
+use crate::{
+    db,
+    models::objects::{StoredObjInfo, StoredOwnerKind},
+    pipeline::{concurrent::Handler, Processor},
+    schema::obj_info,
+};
+
+pub struct ObjInfo;
+
+impl Processor for ObjInfo {
+    const NAME: &'static str = "obj_info";
+    type Value = StoredObjInfo;
+
+    fn process(&self, checkpoint: &Arc<CheckpointData>) -> Result<Vec<Self::Value>> {
+        let cp_sequence_number = checkpoint.checkpoint_summary.sequence_number as i64;
+        let checkpoint_input_objects = checkpoint.checkpoint_input_objects();
+        let latest_live_output_objects = checkpoint
+            .latest_live_output_objects()
+            .into_iter()
+            .map(|o| (o.id(), o))
+            .collect::<BTreeMap<_, _>>();
+        let mut values: BTreeMap<ObjectID, Self::Value> = BTreeMap::new();
+        for object_id in checkpoint_input_objects.keys() {
+            if !latest_live_output_objects.contains_key(object_id) {
+                // If an input object is not in the latest live output objects, it must have been deleted
+                // or wrapped in this checkpoint. We keep an entry for it in the table.
+                // This is necessary when we query objects and iterating over them, so that we don't
+                // include the object in the result if it was deleted.
+                values.insert(
+                    *object_id,
+                    StoredObjInfo {
+                        object_id: object_id.to_vec(),
+                        cp_sequence_number,
+                        owner_kind: None,
+                        owner_id: None,
+                        package: None,
+                        module: None,
+                        name: None,
+                        instantiation: None,
+                    },
+                );
+            }
+        }
+        for (object_id, object) in latest_live_output_objects.iter() {
+            // If an object is newly created/unwrapped in this checkpoint, or if the owner changed,
+            // we need to insert an entry for it in the table.
+            let should_insert = match checkpoint_input_objects.get(object_id) {
+                Some(input_object) => input_object.owner() != object.owner(),
+                None => true,
+            };
+            if should_insert {
+                let type_ = object.type_();
+                values.insert(
+                    *object_id,
+                    StoredObjInfo {
+                        object_id: object_id.to_vec(),
+                        cp_sequence_number,
+                        owner_kind: Some(match object.owner() {
+                            Owner::AddressOwner(_) => StoredOwnerKind::Address,
+                            Owner::ObjectOwner(_) => StoredOwnerKind::Object,
+                            Owner::Shared { .. } => StoredOwnerKind::Shared,
+                            Owner::Immutable => StoredOwnerKind::Immutable,
+                            Owner::ConsensusV2 { .. } => todo!(),
+                        }),
+
+                        owner_id: match object.owner() {
+                            Owner::AddressOwner(a) => Some(a.to_vec()),
+                            Owner::ObjectOwner(o) => Some(o.to_vec()),
+                            Owner::Shared { .. } | Owner::Immutable { .. } => None,
+                            Owner::ConsensusV2 { .. } => todo!(),
+                        },
+
+                        package: type_.map(|t| t.address().to_vec()),
+                        module: type_.map(|t| t.module().to_string()),
+                        name: type_.map(|t| t.name().to_string()),
+                        instantiation: type_
+                            .map(|t| bcs::to_bytes(&t.type_params()))
+                            .transpose()
+                            .map_err(|e| {
+                            anyhow!(
+                                "Failed to serialize type parameters for {}: {e}",
+                                object.id().to_canonical_display(/* with_prefix */ true),
+                            )
+                        })?,
+                    },
+                );
+            }
+        }
+
+        Ok(values.into_values().collect())
+    }
+}
+
+#[async_trait::async_trait]
+impl Handler for ObjInfo {
+    async fn commit(values: &[Self::Value], conn: &mut db::Connection<'_>) -> Result<usize> {
+        Ok(diesel::insert_into(obj_info::table)
+            .values(values)
+            .on_conflict_do_nothing()
+            .execute(conn)
+            .await?)
+    }
+}

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

@@ -11,8 +11,8 @@ use handlers::{
     ev_emit_mod::EvEmitMod, ev_struct_inst::EvStructInst, kv_checkpoints::KvCheckpoints,
     kv_epoch_ends::KvEpochEnds, kv_epoch_starts::KvEpochStarts, kv_feature_flags::KvFeatureFlags,
     kv_objects::KvObjects, kv_protocol_configs::KvProtocolConfigs, kv_transactions::KvTransactions,
-    obj_versions::ObjVersions, sum_coin_balances::SumCoinBalances, sum_displays::SumDisplays,
-    sum_obj_types::SumObjTypes, sum_packages::SumPackages,
+    obj_info::ObjInfo, obj_versions::ObjVersions, sum_coin_balances::SumCoinBalances,
+    sum_displays::SumDisplays, sum_obj_types::SumObjTypes, sum_packages::SumPackages,
     tx_affected_addresses::TxAffectedAddresses, tx_affected_objects::TxAffectedObjects,
     tx_balance_changes::TxBalanceChanges, tx_calls::TxCalls, tx_digests::TxDigests,
     tx_kinds::TxKinds, wal_coin_balances::WalCoinBalances, wal_obj_types::WalObjTypes,
@@ -423,6 +423,7 @@ pub async fn start_indexer(
         kv_objects,
         kv_protocol_configs,
         kv_transactions,
+        obj_info,
         obj_versions,
         tx_affected_addresses,
         tx_affected_objects,
@@ -576,6 +577,7 @@ pub async fn start_indexer(
     add_concurrent!(KvEpochStarts, kv_epoch_starts);
     add_concurrent!(KvObjects, kv_objects);
     add_concurrent!(KvTransactions, kv_transactions);
+    add_concurrent!(ObjInfo, obj_info);
     add_concurrent!(ObjVersions, obj_versions);
     add_concurrent!(TxAffectedAddresses, tx_affected_addresses);
     add_concurrent!(TxAffectedObjects, tx_affected_objects);

crates/sui-indexer-alt/src/models/objects.rs

@@ -9,7 +9,8 @@ use sui_field_count::FieldCount;
 use sui_types::base_types::ObjectID;
 
 use crate::schema::{
-    kv_objects, obj_versions, sum_coin_balances, sum_obj_types, wal_coin_balances, wal_obj_types,
+    kv_objects, obj_info, obj_versions, sum_coin_balances, sum_obj_types, wal_coin_balances,
+    wal_obj_types,
 };
 
 #[derive(Insertable, Debug, Clone, FieldCount)]
@@ -126,3 +127,16 @@ where
         })
     }
 }
+
+#[derive(Insertable, Debug, Clone, FieldCount)]
+#[diesel(table_name = obj_info, primary_key(object_id, cp_sequence_number))]
+pub struct StoredObjInfo {
+    pub object_id: Vec<u8>,
+    pub cp_sequence_number: i64,
+    pub owner_kind: Option<StoredOwnerKind>,
+    pub owner_id: Option<Vec<u8>>,
+    pub package: Option<Vec<u8>>,
+    pub module: Option<String>,
+    pub name: Option<String>,
+    pub instantiation: Option<Vec<u8>>,
+}

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

@@ -103,6 +103,19 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    obj_info (object_id, cp_sequence_number) {
+        object_id -> Bytea,
+        cp_sequence_number -> Int8,
+        owner_kind -> Nullable<Int2>,
+        owner_id -> Nullable<Bytea>,
+        package -> Nullable<Bytea>,
+        module -> Nullable<Text>,
+        name -> Nullable<Text>,
+        instantiation -> Nullable<Bytea>,
+    }
+}
+
 diesel::table! {
     obj_versions (object_id, object_version) {
         object_id -> Bytea,
@@ -250,6 +263,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     kv_objects,
     kv_protocol_configs,
     kv_transactions,
+    obj_info,
     obj_versions,
     sum_coin_balances,
     sum_displays,

crates/sui-types/src/full_checkpoint_content.rs

@@ -1,9 +1,9 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashSet};
 
-use crate::base_types::ObjectRef;
+use crate::base_types::{ObjectID, ObjectRef};
 use crate::effects::{
     IDOperation, ObjectIn, ObjectOut, TransactionEffects, TransactionEffectsAPI, TransactionEvents,
 };
@@ -51,11 +51,24 @@ impl CheckpointData {
         eventually_removed_object_refs.into_values().collect()
     }
 
-    pub fn input_objects(&self) -> Vec<&Object> {
-        self.transactions
-            .iter()
-            .flat_map(|tx| &tx.input_objects)
-            .collect()
+    /// Returns all objects that are used as input to the transactions in the checkpoint,
+    /// and already exist prior to the checkpoint.
+    pub fn checkpoint_input_objects(&self) -> BTreeMap<ObjectID, &Object> {
+        let mut output_objects_seen = HashSet::new();
+        let mut checkpoint_input_objects = BTreeMap::new();
+        for tx in self.transactions.iter() {
+            for obj in tx.input_objects.iter() {
+                let id = obj.id();
+                if output_objects_seen.contains(&id) || checkpoint_input_objects.contains_key(&id) {
+                    continue;
+                }
+                checkpoint_input_objects.insert(id, obj);
+            }
+            for obj in tx.output_objects.iter() {
+                output_objects_seen.insert(obj.id());
+            }
+        }
+        checkpoint_input_objects
     }
 
     pub fn all_objects(&self) -> Vec<&Object> {
@@ -73,7 +86,7 @@ pub struct CheckpointTransaction {
     pub transaction: Transaction,
     /// The effects produced by executing this transaction
     pub effects: TransactionEffects,
-    /// The events, if any, emitted by this transaciton during execution
+    /// The events, if any, emitted by this transactions during execution
     pub events: Option<TransactionEvents>,
     /// The state of all inputs to this transaction as they were prior to execution.
     pub input_objects: Vec<Object>,
@@ -87,7 +100,7 @@ impl CheckpointTransaction {
         // Iterator over id and versions for all deleted or wrapped objects
         match &self.effects {
             TransactionEffects::V1(v1) => Either::Left(
-                // Effects v1 has delted and wrapped objects versions as the "new" version, not the
+                // Effects v1 has deleted and wrapped objects versions as the "new" version, not the
                 // old one that was actually removed. So we need to take these and then look them
                 // up in the `modified_at_versions`.
                 // No need to chain unwrapped_then_deleted because these objects must have been wrapped