Update purge_special_columns to only purge columns that exist

For validators that don't have these columns at all, bail out early.
Otherwise, only purge the primary indexes that are known to exist to
avoid bloating the write batch with extra deletes.

ledger/src/blockstore/blockstore_purge.rs

@@ -335,6 +335,38 @@ impl Blockstore {
         from_slot: Slot,
         to_slot: Slot,
     ) -> Result<()> {
+        let primary_indexes_to_purge: Vec<_> = VALID_LEGACY_PRIMARY_INDEX_VALUES
+            .into_iter()
+            .filter(|index| {
+                let address_signature_present = self
+                    .address_signatures_cf
+                    .iter(IteratorMode::From(
+                        cf::AddressSignatures::as_index(*index),
+                        IteratorDirection::Forward,
+                    ))
+                    .unwrap() // TODO: no unwrap
+                    .next()
+                    .map(|((key_index, _, _, _), _)| key_index == *index)
+                    .unwrap_or(false);
+                let transaction_status_present = self
+                    .transaction_status_cf
+                    .iter(IteratorMode::From(
+                        cf::TransactionStatus::as_index(*index),
+                        IteratorDirection::Forward,
+                    ))
+                    .unwrap() // TODO: no unwrap
+                    .next()
+                    .map(|((key_index, _, _), _)| key_index == *index)
+                    .unwrap_or(false);
+
+                address_signature_present || transaction_status_present
+            })
+            .collect();
+        if primary_indexes_to_purge.is_empty() {
+            // Special columns are empty, nothing to clean so bail early
+            return Ok(());
+        }
+
         let mut index0 = self.transaction_status_index_cf.get(0)?.unwrap_or_default();
         let mut index1 = self.transaction_status_index_cf.get(1)?.unwrap_or_default();
         let to_slot = to_slot.saturating_add(1);
@@ -345,8 +377,9 @@ impl Blockstore {
                 .flat_map(|entry| entry.transactions);
             for transaction in transactions {
                 if let Some(&signature) = transaction.signatures.get(0) {
-                    batch.delete::<cf::TransactionStatus>((0, signature, slot))?;
-                    batch.delete::<cf::TransactionStatus>((1, signature, slot))?;
+                    for index in primary_indexes_to_purge.iter() {
+                        batch.delete::<cf::TransactionStatus>((*index, signature, slot))?;
+                    }
 
                     let meta = self.read_transaction_status((signature, slot))?;
                     let loaded_addresses = meta.map(|meta| meta.loaded_addresses);
@@ -356,8 +389,11 @@ impl Blockstore {
                     );
 
                     for pubkey in account_keys.iter() {
-                        batch.delete::<cf::AddressSignatures>((0, *pubkey, slot, signature))?;
-                        batch.delete::<cf::AddressSignatures>((1, *pubkey, slot, signature))?;
+                        for index in primary_indexes_to_purge.iter() {
+                            batch.delete::<cf::AddressSignatures>((
+                                *index, *pubkey, slot, signature,
+                            ))?;
+                        }
                     }
                 }
             }