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

storage-bigtable/src/bigtable.rs

@@ -683,14 +683,17 @@ impl<F: FnMut(Request<()>) -> InterceptedRequestResult> BigTable<F> {
     pub async fn get_protobuf_or_bincode_cells<'a, B, P>(
         &mut self,
         table: &'a str,
-        row_keys: impl Iterator<Item = String>,
+        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.collect::<Vec<RowKey>>().as_slice())
+            .get_multi_row_data(
+                table,
+                row_keys.into_iter().collect::<Vec<RowKey>>().as_slice(),
+            )
             .await?
             .into_iter()
             .map(|(key, row_data)| {

storage-bigtable/src/lib.rs

@@ -456,10 +456,10 @@ impl LedgerStorage {
     }
 
     // Fetches and gets a vector of confirmed blocks via a multirow fetch
-    pub async fn get_confirmed_blocks_with_data(
+    pub async fn get_confirmed_blocks_with_data<'a>(
         &self,
-        slots: &[Slot],
-    ) -> Result<Vec<(Slot, ConfirmedBlock)>> {
+        slots: &'a [Slot],
+    ) -> Result<impl Iterator<Item = (Slot, ConfirmedBlock)> + 'a> {
         debug!(
             "LedgerStorage::get_confirmed_blocks_with_data request received: {:?}",
             slots
@@ -467,7 +467,7 @@ impl LedgerStorage {
         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: Vec<(Slot, ConfirmedBlock)> = bigtable
+        let data = bigtable
             .get_protobuf_or_bincode_cells("blocks", row_keys)
             .await?
             .filter_map(
@@ -481,8 +481,7 @@ impl LedgerStorage {
                     };
                     Some((key_to_slot(&row_key).unwrap(), block))
                 },
-            )
-            .collect();
+            );
         Ok(data)
     }