[Indexer] Create objects_version table.

## Description

This table maps an object's ID and version to a checkpoint sequence
number, in a table partitioned by the first byte of the object ID.
This speeds up look ups into `objects_history` by offering a path for
a first look-up to the correct partition in that table for a given
object's ID and version.

This PR introduces the table, and the logic to populate it in the
indexer.

## Test plan

```
sui$ cargo nextest run -p sui-indexer
sui$ cargo nextest run -p sui-graphql-rpc
sui$ cargo nextest run -p sui-graphql-e2e-tests --features pg_integration
```

crates/sui-indexer/migrations/mysql/2024-05-05-155158_obj_indices/down.sql

@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+-- TODO: objects_version

crates/sui-indexer/migrations/mysql/2024-05-05-155158_obj_indices/up.sql

@@ -0,0 +1 @@
+-- TODO: objects_version

crates/sui-indexer/migrations/pg/2024-05-05-155158_obj_indices/down.sql

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

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

@@ -5,6 +5,7 @@ pub mod checkpoints;
 pub mod display;
 pub mod epoch;
 pub mod events;
+pub mod obj_indices;
 pub mod objects;
 pub mod packages;
 pub mod transactions;

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

@@ -0,0 +1,37 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use diesel::prelude::*;
+
+use crate::schema::objects_version;
+
+use super::objects::StoredDeletedObject;
+use super::objects::StoredObject;
+
+#[derive(Queryable, Insertable, Debug, Identifiable, Clone, QueryableByName)]
+#[diesel(table_name = objects_version, primary_key(object_id, object_version))]
+pub struct StoredObjectVersion {
+    pub object_id: Vec<u8>,
+    pub object_version: i64,
+    pub cp_sequence_number: i64,
+}
+
+impl From<&StoredObject> for StoredObjectVersion {
+    fn from(o: &StoredObject) -> Self {
+        Self {
+            object_id: o.object_id.clone(),
+            object_version: o.object_version,
+            cp_sequence_number: o.checkpoint_sequence_number,
+        }
+    }
+}
+
+impl From<&StoredDeletedObject> for StoredObjectVersion {
+    fn from(o: &StoredDeletedObject) -> Self {
+        Self {
+            object_id: o.object_id.clone(),
+            object_version: o.object_version,
+            cp_sequence_number: o.checkpoint_sequence_number,
+        }
+    }
+}

crates/sui-indexer/src/schema.rs

@@ -183,6 +183,14 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    objects_version (object_id, object_version) {
+        object_id -> Bytea,
+        object_version -> Int8,
+        cp_sequence_number -> Int8,
+    }
+}
+
 diesel::table! {
     packages (package_id) {
         package_id -> Bytea,
@@ -282,6 +290,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     objects_history,
     objects_history_partition_0,
     objects_snapshot,
+    objects_version,
     packages,
     transactions,
     transactions_partition_0,

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

@@ -31,16 +31,17 @@ use crate::models::checkpoints::StoredCheckpoint;
 use crate::models::display::StoredDisplay;
 use crate::models::epoch::StoredEpochInfo;
 use crate::models::events::StoredEvent;
+use crate::models::obj_indices::StoredObjectVersion;
 use crate::models::objects::{
     StoredDeletedHistoryObject, StoredDeletedObject, StoredHistoryObject, StoredObject,
     StoredObjectSnapshot,
 };
 use crate::models::packages::StoredPackage;
 use crate::models::transactions::StoredTransaction;
 use crate::schema::{
-    checkpoints, display, epochs, events, objects, objects_history, objects_snapshot, packages,
-    transactions, tx_calls, tx_changed_objects, tx_digests, tx_input_objects, tx_recipients,
-    tx_senders,
+    checkpoints, display, epochs, events, objects, objects_history, objects_snapshot,
+    objects_version, packages, transactions, tx_calls, tx_changed_objects, tx_digests,
+    tx_input_objects, tx_recipients, tx_senders,
 };
 use crate::types::{IndexedCheckpoint, IndexedEvent, IndexedPackage, IndexedTransaction, TxIndex};
 use crate::{
@@ -437,13 +438,16 @@ impl<T: R2D2Connection + 'static> PgIndexerStore<T> {
             .checkpoint_db_commit_latency_objects_history_chunks
             .start_timer();
         let mut mutated_objects: Vec<StoredHistoryObject> = vec![];
+        let mut object_versions: Vec<StoredObjectVersion> = vec![];
         let mut deleted_object_ids: Vec<StoredDeletedHistoryObject> = vec![];
         for object in objects {
             match object {
                 ObjectChangeToCommit::MutatedObject(stored_object) => {
+                    object_versions.push(StoredObjectVersion::from(&stored_object));
                     mutated_objects.push(stored_object.into());
                 }
                 ObjectChangeToCommit::DeletedObject(stored_deleted_object) => {
+                    object_versions.push(StoredObjectVersion::from(&stored_deleted_object));
                     deleted_object_ids.push(stored_deleted_object.into());
                 }
             }
@@ -462,6 +466,11 @@ impl<T: R2D2Connection + 'static> PgIndexerStore<T> {
                     );
                 }
 
+                for object_version_chunk in object_versions.chunks(PG_COMMIT_CHUNK_SIZE_INTRA_DB_TX)
+                {
+                    insert_or_ignore_into!(objects_version::table, object_version_chunk, conn);
+                }
+
                 for deleted_objects_chunk in
                     deleted_object_ids.chunks(PG_COMMIT_CHUNK_SIZE_INTRA_DB_TX)
                 {