removes redundant collect::<Vec<_>> in Rocks::multi_get_cf (#1184)

Rocks::multi_get_cf collects an iterator into a vector:
https://github.com/anza-xyz/agave/blob/6b45dc95d/ledger/src/blockstore_db.rs#L661-L666

which is then immediately converted back into an iterator:
https://github.com/anza-xyz/agave/blob/6b45dc95d/ledger/src/blockstore_db.rs#L1589-L1590
https://github.com/anza-xyz/agave/blob/6b45dc95d/ledger/src/blockstore_db.rs#L1705-L1706

The redundant collect::<Vec<_>> can be avoided by returning an iterator
from multi_get_cf.

ledger/src/blockstore.rs

@@ -5377,7 +5377,7 @@ pub mod tests {
         for (i, key) in keys.iter_mut().enumerate().take(TEST_PUT_ENTRY_COUNT) {
             *key = u64::try_from(i).unwrap();
         }
-        let values = blockstore.meta_cf.multi_get(keys.into_iter());
+        let values = blockstore.meta_cf.multi_get(keys);
         for (i, value) in values.iter().enumerate().take(TEST_PUT_ENTRY_COUNT) {
             let k = u64::try_from(i).unwrap();
             assert_eq!(

ledger/src/blockstore_db.rs

@@ -650,21 +650,15 @@ impl Rocks {
         &self,
         cf: &ColumnFamily,
         keys: I,
-    ) -> Vec<Result<Option<DBPinnableSlice>>>
+    ) -> impl Iterator<Item = Result<Option<DBPinnableSlice>>>
     where
         K: AsRef<[u8]> + 'a + ?Sized,
         I: IntoIterator<Item = &'a K>,
     {
-        let values = self
-            .db
-            .batched_multi_get_cf(cf, keys, false)
+        self.db
+            .batched_multi_get_cf(cf, keys, /*sorted_input:*/ false)
             .into_iter()
-            .map(|result| match result {
-                Ok(opt) => Ok(opt),
-                Err(e) => Err(BlockstoreError::RocksDb(e)),
-            })
-            .collect::<Vec<_>>();
-        values
+            .map(|out| out.map_err(BlockstoreError::RocksDb))
     }
 
     fn delete_cf(&self, cf: &ColumnFamily, key: &[u8]) -> Result<()> {
@@ -1574,27 +1568,20 @@ where
         result
     }
 
-    pub fn multi_get_bytes(
-        &self,
-        keys: impl Iterator<Item = C::Index>,
-    ) -> Vec<Result<Option<Vec<u8>>>> {
-        let rocks_keys: Vec<_> = keys.map(|key| C::key(key)).collect();
+    pub(crate) fn multi_get_bytes<I>(&self, keys: I) -> Vec<Result<Option<Vec<u8>>>>
+    where
+        I: IntoIterator<Item = C::Index>,
+    {
+        let keys: Vec<_> = keys.into_iter().map(C::key).collect();
         {
             let is_perf_enabled = maybe_enable_rocksdb_perf(
                 self.column_options.rocks_perf_sample_interval,
                 &self.read_perf_status,
             );
             let result = self
                 .backend
-                .multi_get_cf(self.handle(), rocks_keys.iter())
-                .into_iter()
-                .map(|r| match r {
-                    Ok(opt) => match opt {
-                        Some(pinnable_slice) => Ok(Some(pinnable_slice.as_ref().to_vec())),
-                        None => Ok(None),
-                    },
-                    Err(e) => Err(e),
-                })
+                .multi_get_cf(self.handle(), &keys)
+                .map(|out| Ok(out?.as_deref().map(<[u8]>::to_vec)))
                 .collect::<Vec<Result<Option<_>>>>();
             if let Some(op_start_instant) = is_perf_enabled {
                 // use multi-get instead
@@ -1693,24 +1680,20 @@ impl<C> LedgerColumn<C>
 where
     C: TypedColumn + ColumnName,
 {
-    pub fn multi_get(&self, keys: impl Iterator<Item = C::Index>) -> Vec<Result<Option<C::Type>>> {
-        let rocks_keys: Vec<_> = keys.map(|key| C::key(key)).collect();
+    pub(crate) fn multi_get<I>(&self, keys: I) -> Vec<Result<Option<C::Type>>>
+    where
+        I: IntoIterator<Item = C::Index>,
+    {
+        let keys: Vec<_> = keys.into_iter().map(C::key).collect();
         {
             let is_perf_enabled = maybe_enable_rocksdb_perf(
                 self.column_options.rocks_perf_sample_interval,
                 &self.read_perf_status,
             );
             let result = self
                 .backend
-                .multi_get_cf(self.handle(), rocks_keys.iter())
-                .into_iter()
-                .map(|r| match r {
-                    Ok(opt) => match opt {
-                        Some(pinnable_slice) => Ok(Some(deserialize(pinnable_slice.as_ref())?)),
-                        None => Ok(None),
-                    },
-                    Err(e) => Err(e),
-                })
+                .multi_get_cf(self.handle(), &keys)
+                .map(|out| Ok(out?.as_deref().map(deserialize).transpose()?))
                 .collect::<Vec<Result<Option<_>>>>();
             if let Some(op_start_instant) = is_perf_enabled {
                 // use multi-get instead