store: Create contract_address index before call cache eviction

Add catalog::table_has_index() to check for index existence, and
use it in clear_stale_call_cache to ensure a btree index on
contract_address exists before running deletion queries. The index
is created concurrently if missing, avoiding sequential scans on
large call_cache tables.

Author: lutter

store/postgres/src/catalog.rs

@@ -913,6 +913,32 @@ pub(crate) async fn indexes_for_table(
     Ok(results.into_iter().map(|i| i.def).collect())
 }
 
+pub(crate) async fn table_has_index(
+    conn: &mut AsyncPgConnection,
+    schema_name: &str,
+    index_name: &str,
+) -> Result<bool, StoreError> {
+    #[derive(QueryableByName)]
+    #[allow(dead_code)]
+    struct Exists {
+        #[diesel(sql_type = diesel::sql_types::Integer)]
+        exists: i32,
+    }
+
+    let exists = sql_query(
+        "SELECT 1 AS exists FROM pg_indexes \
+         WHERE schemaname = $1 AND indexname = $2",
+    )
+    .bind::<Text, _>(schema_name)
+    .bind::<Text, _>(index_name)
+    .get_result::<Exists>(conn)
+    .await
+    .optional()
+    .map_err::<StoreError, _>(Into::into)?;
+
+    Ok(exists.is_some())
+}
+
 pub(crate) async fn drop_index(
     conn: &mut AsyncPgConnection,
     schema_name: &str,

store/postgres/src/chain_store.rs

@@ -1643,13 +1643,68 @@ mod data {
             }
         }
 
+        const CALL_CACHE_CONTRACT_ADDRESS_INDEX: &str = "call_cache_contract_address";
+
+        /// Ensure that an index on `contract_address` exists on the
+        /// call_cache table to speed up deletion queries. If the index does
+        /// not exist, create it concurrently.
+        async fn ensure_contract_address_index(
+            &self,
+            conn: &mut AsyncPgConnection,
+            logger: &Logger,
+        ) -> Result<(), Error> {
+            let (schema_name, table_qname) = match self {
+                Storage::Shared => ("public", "public.eth_call_cache".to_string()),
+                Storage::Private(Schema {
+                    name, call_cache, ..
+                }) => (name.as_str(), call_cache.qname.clone()),
+            };
+
+            let has_index = crate::catalog::table_has_index(
+                conn,
+                schema_name,
+                Self::CALL_CACHE_CONTRACT_ADDRESS_INDEX,
+            )
+            .await?;
+
+            if !has_index {
+                let start = Instant::now();
+                info!(
+                    logger,
+                    "Creating index {} on {}.contract_address; \
+                     this may take a long time",
+                    Self::CALL_CACHE_CONTRACT_ADDRESS_INDEX,
+                    table_qname
+                );
+                conn.batch_execute(&format!(
+                    "create index concurrently if not exists {} \
+                     on {}(contract_address)",
+                    Self::CALL_CACHE_CONTRACT_ADDRESS_INDEX,
+                    table_qname
+                ))
+                .await?;
+                let duration = start.elapsed();
+                info!(
+                    logger,
+                    "Finished creating index {} on {}.contract_address in {:?}",
+                    Self::CALL_CACHE_CONTRACT_ADDRESS_INDEX,
+                    table_qname,
+                    duration
+                );
+            }
+
+            Ok(())
+        }
+
         pub async fn clear_stale_call_cache(
             &self,
             conn: &mut AsyncPgConnection,
             logger: &Logger,
             ttl_days: i32,
             ttl_max_contracts: Option<i64>,
         ) -> Result<(), Error> {
+            self.ensure_contract_address_index(conn, logger).await?;
+
             let mut total_calls: usize = 0;
             let mut total_contracts: i64 = 0;
             // We process contracts in batches to avoid loading too many