Make Blockstore::purge_special_columns_exact() bail if columns empty (#33534)

The special columns, TransactionStatus and AddressSignatures, are only
populated if --enable-rpc-transaction-history is passed. Cleaning these
columns for a range of slots is very expensive, as the block for each
slot must be read, deserialized, and then parsed to extract all of the
transaction signatures and address pubkeys.

This change adds a simple check to see if there are any values at all in
the special columns. If there are not, then the whole process described
above can be skipped for nodes that are not storing the special columns.

ledger/src/blockstore/blockstore_purge.rs

@@ -339,6 +339,26 @@ impl Blockstore {
                 .is_ok()
     }
 
+    /// Returns true if the special columns, TransactionStatus and
+    /// AddressSignatures, are both empty.
+    ///
+    /// It should not be the case that one is empty and the other is not, but
+    /// just return false in this case.
+    fn special_columns_empty(&self) -> Result<bool> {
+        let transaction_status_empty = self
+            .transaction_status_cf
+            .iter(IteratorMode::Start)?
+            .next()
+            .is_none();
+        let address_signatures_empty = self
+            .address_signatures_cf
+            .iter(IteratorMode::Start)?
+            .next()
+            .is_none();
+
+        Ok(transaction_status_empty && address_signatures_empty)
+    }
+
     /// Purges special columns (using a non-Slot primary-index) exactly, by
     /// deserializing each slot being purged and iterating through all
     /// transactions to determine the keys of individual records.
@@ -352,6 +372,10 @@ impl Blockstore {
         from_slot: Slot,
         to_slot: Slot,
     ) -> Result<()> {
+        if self.special_columns_empty()? {
+            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 slot_indexes = |slot: Slot| -> Vec<u64> {
@@ -859,6 +883,54 @@ pub mod tests {
         );
     }
 
+    #[test]
+    fn test_special_columns_empty() {
+        let ledger_path = get_tmp_ledger_path_auto_delete!();
+        let blockstore = Blockstore::open(ledger_path.path()).unwrap();
+
+        // Nothing has been inserted yet
+        assert!(blockstore.special_columns_empty().unwrap());
+
+        let num_entries = 1;
+        let max_slot = 9;
+        for slot in 0..=max_slot {
+            let entries = make_slot_entries_with_transactions(num_entries);
+            let shreds = entries_to_test_shreds(
+                &entries,
+                slot,
+                slot.saturating_sub(1),
+                true, // is_full_slot
+                0,    // version
+                true, // merkle_variant
+            );
+            blockstore.insert_shreds(shreds, None, false).unwrap();
+
+            for transaction in entries.into_iter().flat_map(|entry| entry.transactions) {
+                assert_eq!(transaction.signatures.len(), 1);
+                blockstore
+                    .write_transaction_status(
+                        slot,
+                        transaction.signatures[0],
+                        transaction.message.static_account_keys().iter().collect(),
+                        vec![],
+                        TransactionStatusMeta::default(),
+                    )
+                    .unwrap();
+            }
+        }
+        assert!(!blockstore.special_columns_empty().unwrap());
+
+        // Partially purge and ensure special columns are non-empty
+        blockstore
+            .run_purge(0, max_slot - 5, PurgeType::Exact)
+            .unwrap();
+        assert!(!blockstore.special_columns_empty().unwrap());
+
+        // Purge the rest and ensure the special columns are empty once again
+        blockstore.run_purge(0, max_slot, PurgeType::Exact).unwrap();
+        assert!(blockstore.special_columns_empty().unwrap());
+    }
+
     #[test]
     #[allow(clippy::cognitive_complexity)]
     fn test_purge_transaction_status_exact() {