Add get_confirmed_blocks_with_data method (backport #23618) (#23939)

* add get_confirmed_blocks_with_data and get_protobuf_or_bincode_cells

(cherry picked from commit f3219fb695)

* appease clippy

(cherry picked from commit 5533e9393c)

* use &[T] instead of Vec<T> where appropriate

clippy

(cherry picked from commit fbcf6a0802)

* modify get_protobuf_or_bincode_cells to accept and return an iterator

(cherry picked from commit f717fda9a3)

* make get_protobuf_or_bincode_cells accept IntoIter on row_keys, make get_confirmed_blocks_with_data return an Iterator

(cherry picked from commit d8be0d9430)

Co-authored-by: Edgar Xi <edgarxi97@gmail.com>

storage-bigtable/src/bigtable.rs

@@ -680,6 +680,32 @@ impl<F: FnMut(Request<()>) -> InterceptedRequestResult> BigTable<F> {
         deserialize_protobuf_or_bincode_cell_data(&row_data, table, key)
     }
 
+    #[allow(clippy::needless_lifetimes)]
+    pub async fn get_protobuf_or_bincode_cells<'a, B, P>(
+        &mut self,
+        table: &'a str,
+        row_keys: impl IntoIterator<Item = RowKey>,
+    ) -> Result<impl Iterator<Item = (RowKey, CellData<B, P>)> + 'a>
+    where
+        B: serde::de::DeserializeOwned,
+        P: prost::Message + Default,
+    {
+        Ok(self
+            .get_multi_row_data(
+                table,
+                row_keys.into_iter().collect::<Vec<RowKey>>().as_slice(),
+            )
+            .await?
+            .into_iter()
+            .map(|(key, row_data)| {
+                let key_str = key.to_string();
+                (
+                    key,
+                    deserialize_protobuf_or_bincode_cell_data(&row_data, table, key_str).unwrap(),
+                )
+            }))
+    }
+
     pub async fn put_bincode_cells<T>(
         &mut self,
         table: &str,

storage-bigtable/src/lib.rs

@@ -1,5 +1,6 @@
 #![allow(clippy::integer_arithmetic)]
 use {
+    crate::bigtable::RowKey,
     log::*,
     serde::{Deserialize, Serialize},
     solana_metrics::inc_new_counter_debug,
@@ -393,6 +394,37 @@ impl LedgerStorage {
         Ok(blocks.into_iter().filter_map(|s| key_to_slot(&s)).collect())
     }
 
+    // Fetches and gets a vector of confirmed blocks via a multirow fetch
+    #[allow(clippy::needless_lifetimes)]
+    pub async fn get_confirmed_blocks_with_data<'a>(
+        &self,
+        slots: &'a [Slot],
+    ) -> Result<impl Iterator<Item = (Slot, ConfirmedBlockWithOptionalMetadata)> + 'a> {
+        debug!(
+            "LedgerStorage::get_confirmed_blocks_with_data request received: {:?}",
+            slots
+        );
+        inc_new_counter_debug!("storage-bigtable-query", 1);
+        let mut bigtable = self.connection.client();
+        let row_keys = slots.iter().copied().map(slot_to_blocks_key);
+        let data = bigtable
+            .get_protobuf_or_bincode_cells("blocks", row_keys)
+            .await?
+            .filter_map(
+                |(row_key, block_cell_data): (
+                    RowKey,
+                    bigtable::CellData<StoredConfirmedBlock, generated::ConfirmedBlock>,
+                )| {
+                    let block = match block_cell_data {
+                        bigtable::CellData::Bincode(block) => block.into(),
+                        bigtable::CellData::Protobuf(block) => block.try_into().ok()?,
+                    };
+                    Some((key_to_slot(&row_key).unwrap(), block))
+                },
+            );
+        Ok(data)
+    }
+
     /// Fetch the confirmed block from the desired slot
     pub async fn get_confirmed_block(
         &self,