Indexer pipeline (#13075)

## Description 

Describe the changes or additions included in this PR.

## Test Plan 

How did you test the new or updated feature?

---
If your changes are not user-facing and not a breaking change, you can
skip the following section. Otherwise, please indicate what changed, and
then add to the Release Notes section as highlighted during the release
process.

### Type of Change (Check all that apply)

- [ ] protocol change
- [ ] user-visible impact
- [ ] breaking change for a client SDKs
- [ ] breaking change for FNs (FN binary must upgrade)
- [ ] breaking change for validators or node operators (must upgrade
binaries)
- [ ] breaking change for on-chain data layout
- [ ] necessitate either a data wipe or data migration

### Release notes

crates/sui-indexer/Cargo.toml

@@ -15,6 +15,7 @@ backoff.workspace = true
 bcs.workspace = true
 chrono.workspace = true
 clap.workspace = true
+tap.workspace = true
 diesel.workspace = true
 diesel-derive-enum.workspace = true
 futures.workspace = true

crates/sui-indexer/benches/indexer_benchmark.rs

@@ -14,21 +14,15 @@ use tokio::runtime::Runtime;
 
 use sui_indexer::metrics::IndexerMetrics;
 use sui_indexer::models::checkpoints::Checkpoint;
-use sui_indexer::models::objects::{NamedBcsBytes, Object as DBObject, ObjectStatus};
-use sui_indexer::models::owners::OwnerType;
 use sui_indexer::models::transactions::Transaction;
 use sui_indexer::new_pg_connection_pool;
-use sui_indexer::store::{
-    IndexerStore, PgIndexerStore, TemporaryCheckpointStore, TransactionObjectChanges,
-};
+use sui_indexer::store::{IndexerStore, PgIndexerStore, TemporaryCheckpointStore};
 use sui_indexer::utils::reset_database;
 use sui_json_rpc_types::CheckpointId;
 use sui_types::base_types::{ObjectDigest, ObjectID, SequenceNumber, SuiAddress};
 use sui_types::crypto::AggregateAuthoritySignature;
 use sui_types::digests::TransactionDigest;
-use sui_types::gas_coin::GasCoin;
 use sui_types::messages_checkpoint::CheckpointDigest;
-use sui_types::object::Object;
 use sui_types::transaction::{TransactionData, TEST_ONLY_GAS_UNIT_FOR_TRANSFER};
 
 fn indexer_benchmark(c: &mut Criterion) {
@@ -83,10 +77,6 @@ fn create_checkpoint(sequence_number: i64) -> TemporaryCheckpointStore {
             .map(|_| create_transaction(sequence_number))
             .collect(),
         events: vec![],
-        object_changes: vec![TransactionObjectChanges {
-            changed_objects: (1..1000).map(|_| create_object(sequence_number)).collect(),
-            deleted_objects: vec![],
-        }],
         packages: vec![],
         input_objects: vec![],
         changed_objects: vec![],
@@ -137,32 +127,6 @@ fn create_transaction(sequence_number: i64) -> Transaction {
     }
 }
 
-fn create_object(sequence_number: i64) -> DBObject {
-    DBObject {
-        epoch: 0,
-        checkpoint: sequence_number,
-        object_id: ObjectID::random().to_string(),
-        version: 0,
-        object_digest: ObjectDigest::random().to_string(),
-        owner_type: OwnerType::AddressOwner,
-        owner_address: Some(SuiAddress::random_for_testing_only().to_string()),
-        initial_shared_version: None,
-        previous_transaction: TransactionDigest::random().base58_encode(),
-        object_type: GasCoin::type_().to_string(),
-        object_status: ObjectStatus::Created,
-        has_public_transfer: true,
-        storage_rebate: 0,
-        bcs: vec![NamedBcsBytes(
-            "data".to_string(),
-            Object::new_gas_for_testing()
-                .data
-                .try_as_move()
-                .unwrap()
-                .contents()
-                .to_vec(),
-        )],
-    }
-}
 criterion_group! {
     name = benches;
     config = Criterion::default().sample_size(50).measurement_time(Duration::from_secs(10));

crates/sui-indexer/src/errors.rs

@@ -9,6 +9,22 @@ use thiserror::Error;
 use sui_types::base_types::ObjectIDParseError;
 use sui_types::error::{SuiError, SuiObjectResponseError, UserInputError};
 
+#[derive(Debug, Error)]
+pub struct DataDownloadError {
+    pub error: IndexerError,
+    pub next_checkpoint_sequence_number: u64,
+}
+
+impl std::fmt::Display for DataDownloadError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "next_checkpoint_seq: {}, error: {}",
+            self.next_checkpoint_sequence_number, self.error
+        )
+    }
+}
+
 #[derive(Debug, Error)]
 pub enum IndexerError {
     #[error("Indexer failed to convert timestamp to NaiveDateTime with error: `{0}`")]