indexer: clean up legacy snapshot codes (#19078)

## Description 

the legacy method of updating `objects_snapshot` is no longer needed,
thus this pr cleans it up

## Test plan 

ci

---

## 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/src/handlers/objects_snapshot_processor.rs

@@ -218,8 +218,7 @@ impl ObjectsSnapshotProcessor {
                             let last_checkpoint_seq = batch.last().as_ref().unwrap().checkpoint_sequence_number;
                             info!("Objects snapshot processor is updating objects snapshot table from {} to {}", first_checkpoint_seq, last_checkpoint_seq);
 
-
-                            store.backfill_objects_snapshot(batch.drain(..).map(|obj| obj.object_changes).collect())
+                            store.persist_objects_snapshot(batch.drain(..).map(|obj| obj.object_changes).collect())
                                 .await
                                 .unwrap_or_else(|_| panic!("Failed to backfill objects snapshot from {} to {}", first_checkpoint_seq, last_checkpoint_seq));
                             start_cp = last_checkpoint_seq + 1;

crates/sui-indexer/src/metrics.rs

@@ -130,7 +130,6 @@ pub struct IndexerMetrics {
     pub epoch_db_commit_latency: Histogram,
     // latencies of slow DB update queries, now only advance epoch and objects_snapshot update
     pub advance_epoch_latency: Histogram,
-    pub update_object_snapshot_latency: Histogram,
     // latencies of RPC endpoints in read.rs
     pub get_transaction_block_latency: Histogram,
     pub multi_get_transaction_blocks_latency: Histogram,
@@ -569,12 +568,6 @@ impl IndexerMetrics {
                 DB_UPDATE_QUERY_LATENCY_SEC_BUCKETS.to_vec(),
                 registry,
             ).unwrap(),
-            update_object_snapshot_latency: register_histogram_with_registry!(
-                "update_object_snapshot_latency",
-                "Time spent in updating object snapshot",
-                DB_UPDATE_QUERY_LATENCY_SEC_BUCKETS.to_vec(),
-                registry,
-            ).unwrap(),
             subscription_process_latency: register_histogram_with_registry!(
                 "subscription_process_latency",
                 "Time spent in process Websocket subscription",

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

@@ -48,16 +48,11 @@ pub trait IndexerStore: Clone + Sync + Send + 'static {
         object_changes: Vec<TransactionObjectChangesToCommit>,
     ) -> Result<(), IndexerError>;
 
-    // persist objects snapshot with object changes during backfill
-    async fn backfill_objects_snapshot(
+    async fn persist_objects_snapshot(
         &self,
         object_changes: Vec<TransactionObjectChangesToCommit>,
     ) -> Result<(), IndexerError>;
 
-    // update objects snapshot after backfill is done
-    async fn update_objects_snapshot(&self, start_cp: u64, end_cp: u64)
-        -> Result<(), IndexerError>;
-
     async fn persist_checkpoints(
         &self,
         checkpoints: Vec<IndexedCheckpoint>,

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

@@ -96,38 +96,6 @@ const PG_COMMIT_PARALLEL_CHUNK_SIZE: usize = 100;
 const PG_COMMIT_OBJECTS_PARALLEL_CHUNK_SIZE: usize = 500;
 const PG_DB_COMMIT_SLEEP_DURATION: Duration = Duration::from_secs(3600);
 
-// with rn = 1, we only select the latest version of each object,
-// so that we don't have to update the same object multiple times.
-const UPDATE_OBJECTS_SNAPSHOT_QUERY: &str = r"
-INSERT INTO objects_snapshot (object_id, object_version, object_status, object_digest, checkpoint_sequence_number, owner_type, owner_id, object_type, object_type_package, object_type_module, object_type_name, serialized_object, coin_type, coin_balance, df_kind, df_name, df_object_type, df_object_id)
-SELECT object_id, object_version, object_status, object_digest, checkpoint_sequence_number, owner_type, owner_id, object_type, object_type_package, object_type_module, object_type_name, serialized_object, coin_type, coin_balance, df_kind, df_name, df_object_type, df_object_id
-FROM (
-    SELECT *,
-           ROW_NUMBER() OVER (PARTITION BY object_id ORDER BY object_version DESC) as rn
-    FROM objects_history
-    WHERE checkpoint_sequence_number >= $1 AND checkpoint_sequence_number < $2
-) as subquery
-WHERE rn = 1
-ON CONFLICT (object_id) DO UPDATE
-SET object_version = EXCLUDED.object_version,
-    object_status = EXCLUDED.object_status,
-    object_digest = EXCLUDED.object_digest,
-    checkpoint_sequence_number = EXCLUDED.checkpoint_sequence_number,
-    owner_type = EXCLUDED.owner_type,
-    owner_id = EXCLUDED.owner_id,
-    object_type = EXCLUDED.object_type,
-    object_type_package = EXCLUDED.object_type_package,
-    object_type_module = EXCLUDED.object_type_module,
-    object_type_name = EXCLUDED.object_type_name,
-    serialized_object = EXCLUDED.serialized_object,
-    coin_type = EXCLUDED.coin_type,
-    coin_balance = EXCLUDED.coin_balance,
-    df_kind = EXCLUDED.df_kind,
-    df_name = EXCLUDED.df_name,
-    df_object_type = EXCLUDED.df_object_type,
-    df_object_id = EXCLUDED.df_object_id;
-";
-
 #[derive(Clone)]
 pub struct PgIndexerStoreConfig {
     pub parallel_chunk_size: usize,
@@ -619,34 +587,6 @@ impl PgIndexerStore {
         })
     }
 
-    fn update_objects_snapshot(&self, start_cp: u64, end_cp: u64) -> Result<(), IndexerError> {
-        let work_mem_gb = std::env::var("INDEXER_PG_WORK_MEM")
-            .unwrap_or_else(|_e| "16".to_string())
-            .parse::<i64>()
-            .unwrap();
-        let pg_work_mem_query_string = format!("SET work_mem = '{}GB'", work_mem_gb);
-        let pg_work_mem_query = pg_work_mem_query_string.as_str();
-        transactional_blocking_with_retry!(
-            &self.blocking_cp,
-            |conn| { RunQueryDsl::execute(diesel::sql_query(pg_work_mem_query), conn,) },
-            PG_DB_COMMIT_SLEEP_DURATION
-        )?;
-
-        transactional_blocking_with_retry!(
-            &self.blocking_cp,
-            |conn| {
-                RunQueryDsl::execute(
-                    diesel::sql_query(UPDATE_OBJECTS_SNAPSHOT_QUERY)
-                        .bind::<diesel::sql_types::BigInt, _>(start_cp as i64)
-                        .bind::<diesel::sql_types::BigInt, _>(end_cp as i64),
-                    conn,
-                )
-            },
-            PG_DB_COMMIT_SLEEP_DURATION
-        )?;
-        Ok(())
-    }
-
     fn persist_checkpoints(&self, checkpoints: Vec<IndexedCheckpoint>) -> Result<(), IndexerError> {
         let Some(first_checkpoint) = checkpoints.first() else {
             return Ok(());
@@ -1764,7 +1704,7 @@ impl IndexerStore for PgIndexerStore {
         Ok(())
     }
 
-    async fn backfill_objects_snapshot(
+    async fn persist_objects_snapshot(
         &self,
         object_changes: Vec<TransactionObjectChangesToCommit>,
     ) -> Result<(), IndexerError> {
@@ -1859,37 +1799,6 @@ impl IndexerStore for PgIndexerStore {
         Ok(())
     }
 
-    async fn update_objects_snapshot(
-        &self,
-        start_cp: u64,
-        end_cp: u64,
-    ) -> Result<(), IndexerError> {
-        let skip_snapshot = std::env::var("SKIP_OBJECT_SNAPSHOT")
-            .map(|val| val.eq_ignore_ascii_case("true"))
-            .unwrap_or(false);
-        if skip_snapshot {
-            info!("skipping object snapshot");
-            return Ok(());
-        }
-
-        let guard = self.metrics.update_object_snapshot_latency.start_timer();
-
-        self.spawn_blocking_task(move |this| this.update_objects_snapshot(start_cp, end_cp))
-            .await
-            .map_err(|e| {
-                IndexerError::PostgresWriteError(format!(
-                    "Failed to update objects snapshot: {:?}",
-                    e
-                ))
-            })??;
-        let elapsed = guard.stop_and_record();
-        info!(
-            elapsed,
-            "Persisted snapshot for checkpoints from {} to {}", start_cp, end_cp
-        );
-        Ok(())
-    }
-
     async fn persist_checkpoints(
         &self,
         checkpoints: Vec<IndexedCheckpoint>,