[Indexer] Create objects_version table. (#17542)

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.

```
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
```

A future PR will make use of this table from GraphQL, which will test it
further.

- #17686
- #17687
- #17688
- #17689
- #17691
- #17694
- #17695

---

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:

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

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

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

@@ -0,0 +1,9 @@
+-- The Postgres version of this table is partitioned by the first byte
+-- of object_id, but this kind of partition is not easily supported in
+-- MySQL, so this variant is unpartitioned for now.
+CREATE TABLE objects_version (
+    object_id           BLOB          NOT NULL,
+    object_version      BIGINT        NOT NULL,
+    cp_sequence_number  BIGINT        NOT NULL,
+    PRIMARY KEY (object_id(32), object_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/migrations/pg/2024-05-05-155158_obj_indices/up.sql

@@ -0,0 +1,31 @@
+-- Indexing table mapping an object's ID and version to its checkpoint
+-- sequence number, partitioned by the first byte of its Object ID.
+CREATE TABLE objects_version (
+    object_id           bytea         NOT NULL,
+    object_version      bigint        NOT NULL,
+    cp_sequence_number  bigint        NOT NULL,
+    PRIMARY KEY (object_id, object_version)
+) PARTITION BY RANGE (object_id);
+
+-- Create a partition for each first byte value.
+DO $$
+DECLARE
+    lo text;
+    hi text;
+BEGIN
+    FOR i IN 0..254 LOOP
+        lo := LPAD(TO_HEX(i), 2, '0');
+        hi := LPAD(TO_HEX(i + 1), 2, '0');
+        EXECUTE FORMAT($F$
+            CREATE TABLE objects_version_%1$s PARTITION OF objects_version FOR VALUES
+            FROM (E'\\x%1$s00000000000000000000000000000000000000000000000000000000000000')
+            TO   (E'\\x%2$s00000000000000000000000000000000000000000000000000000000000000');
+        $F$, lo, hi);
+    END LOOP;
+END;
+$$ LANGUAGE plpgsql;
+
+-- Special case for the last partition, because of the upper bound.
+CREATE TABLE objects_version_ff PARTITION OF objects_version FOR VALUES
+FROM (E'\\xff00000000000000000000000000000000000000000000000000000000000000')
+TO   (MAXVALUE);

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,40 @@
+// 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;
+
+/// Model types related to tables that support efficient execution of queries on the `objects`,
+/// `objects_history` and `objects_snapshot` tables.
+
+#[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/mod.rs

@@ -20,6 +20,7 @@ mod inner {
     pub use crate::schema::pg::objects;
     pub use crate::schema::pg::objects_history;
     pub use crate::schema::pg::objects_snapshot;
+    pub use crate::schema::pg::objects_version;
     pub use crate::schema::pg::packages;
     pub use crate::schema::pg::pruner_cp_watermark;
     pub use crate::schema::pg::transactions;
@@ -42,6 +43,7 @@ mod inner {
     pub use crate::schema::mysql::objects;
     pub use crate::schema::mysql::objects_history;
     pub use crate::schema::mysql::objects_snapshot;
+    pub use crate::schema::mysql::objects_version;
     pub use crate::schema::mysql::packages;
     pub use crate::schema::mysql::pruner_cp_watermark;
     pub use crate::schema::mysql::transactions;
@@ -61,6 +63,7 @@ pub use inner::events;
 pub use inner::objects;
 pub use inner::objects_history;
 pub use inner::objects_snapshot;
+pub use inner::objects_version;
 pub use inner::packages;
 pub use inner::pruner_cp_watermark;
 pub use inner::transactions;

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

@@ -148,6 +148,14 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    objects_version (object_id, object_version) {
+        object_id -> Blob,
+        object_version -> Bigint,
+        cp_sequence_number -> Bigint,
+    }
+}
+
 diesel::table! {
     packages (package_id) {
         package_id -> Blob,

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

@@ -197,6 +197,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,
@@ -299,6 +307,7 @@ macro_rules! for_all_tables {
             objects,
             objects_history,
             objects_snapshot,
+            objects_version,
             packages,
             transactions,
             tx_calls,

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

@@ -33,6 +33,7 @@ use crate::models::checkpoints::StoredCpTx;
 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,
@@ -41,8 +42,8 @@ use crate::models::packages::StoredPackage;
 use crate::models::transactions::StoredTransaction;
 use crate::schema::{
     chain_identifier, checkpoints, display, epochs, events, objects, objects_history,
-    objects_snapshot, packages, pruner_cp_watermark, transactions, tx_calls, tx_changed_objects,
-    tx_digests, tx_input_objects, tx_recipients, tx_senders,
+    objects_snapshot, objects_version, packages, pruner_cp_watermark, 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::{
@@ -522,13 +523,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());
                 }
             }
@@ -547,6 +551,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)
                 {