switch to HashMap for block indexing

pageserver/src/layered_repository/delta_layer.rs

@@ -391,7 +391,7 @@ impl DeltaLayer {
         end_lsn: Lsn,
         dropped: bool,
         predecessor: Option<Arc<dyn Layer>>,
-        page_versions: impl Iterator<Item = (&'a u32, &'a Lsn, &'a PageVersion)>,
+        page_versions: impl Iterator<Item = (u32, Lsn, &'a PageVersion)>,
         relsizes: OrderedVec<Lsn, u32>,
     ) -> Result<DeltaLayer> {
         let delta_layer = DeltaLayer {
@@ -428,7 +428,7 @@ impl DeltaLayer {
             let buf = PageVersion::ser(page_version)?;
             let blob_range = page_version_writer.write_blob(&buf)?;
 
-            inner.page_version_metas.append((*blknum, *lsn), blob_range);
+            inner.page_version_metas.append((blknum, lsn), blob_range);
         }
 
         let book = page_version_writer.close()?;

pageserver/src/layered_repository/inmemory_layer.rs

@@ -16,7 +16,7 @@ use anyhow::{bail, Result};
 use bytes::Bytes;
 use log::*;
 use std::cmp::min;
-use std::collections::BTreeMap;
+use std::collections::HashMap;
 use std::ops::Bound::Included;
 use std::ops::RangeBounds;
 use std::path::PathBuf;
@@ -50,7 +50,7 @@ pub struct InMemoryLayer {
 }
 
 #[derive(Default)]
-struct PageVersions(BTreeMap<u32, OrderedVec<Lsn, PageVersion>>);
+struct PageVersions(HashMap<u32, OrderedVec<Lsn, PageVersion>>);
 
 impl PageVersions {
     fn range<R: RangeBounds<Lsn>>(&self, blknum: u32, lsn_range: R) -> &[(Lsn, PageVersion)] {
@@ -65,15 +65,24 @@ impl PageVersions {
         ordered_vec.append_update(lsn, page_version);
     }
 
-    fn iter(&self) -> PageVersionIter {
-        let mut map_iter = self.0.iter();
-        let map_entry = map_iter
-            .next()
-            .map(|(blknum, ordered_vec)| (blknum, ordered_vec.iter()));
+    fn iter(&self) -> OrderedBlockIter {
+        let mut block_numbers: Vec<u32> = self.0.keys().cloned().collect();
+        // TODO consider counting sort given the small size of the key space
+        block_numbers.sort_unstable();
 
-        PageVersionIter {
-            map_iter,
-            map_entry,
+        let cur_idx = 0;
+        let ordered_vec_iter = block_numbers
+            .get(cur_idx)
+            .map(|blk_num| self.0.get(blk_num).unwrap().iter())
+            .unwrap_or_else(|| [].iter());
+
+        OrderedBlockIter {
+            page_versions: self,
+
+            block_numbers,
+            cur_idx,
+
+            ordered_vec_iter,
         }
     }
 
@@ -82,25 +91,28 @@ impl PageVersions {
     }
 }
 
-struct PageVersionIter<'a> {
-    map_iter: std::collections::btree_map::Iter<'a, u32, OrderedVec<Lsn, PageVersion>>,
-    map_entry: Option<(&'a u32, std::slice::Iter<'a, (Lsn, PageVersion)>)>,
+struct OrderedBlockIter<'a> {
+    page_versions: &'a PageVersions,
+
+    block_numbers: Vec<u32>,
+    cur_idx: usize,
+
+    ordered_vec_iter: std::slice::Iter<'a, (Lsn, PageVersion)>,
 }
 
-impl<'a> Iterator for PageVersionIter<'a> {
-    type Item = (&'a u32, &'a Lsn, &'a PageVersion);
+impl<'a> Iterator for OrderedBlockIter<'a> {
+    type Item = (u32, Lsn, &'a PageVersion);
 
     fn next(&mut self) -> Option<Self::Item> {
         loop {
-            let (blknum, iter) = self.map_entry.as_mut()?;
-            if let Some((lsn, pv)) = iter.next() {
-                return Some((blknum, lsn, pv));
+            if let Some((lsn, page_version)) = self.ordered_vec_iter.next() {
+                let blk_num = self.block_numbers[self.cur_idx];
+                return Some((blk_num, *lsn, page_version));
             }
 
-            self.map_entry = self
-                .map_iter
-                .next()
-                .map(|(blknum, ordered_vec)| (blknum, ordered_vec.iter()));
+            let blk_num = self.block_numbers.get(self.cur_idx + 1)?;
+            self.cur_idx += 1;
+            self.ordered_vec_iter = self.page_versions.0.get(blk_num).unwrap().iter();
         }
     }
 }
@@ -763,7 +775,7 @@ impl InMemoryLayer {
         let mut before_page_versions = inner.page_versions.iter().filter(|&tup| {
             let (_blknum, lsn, _pv) = tup;
 
-            *lsn < end_lsn
+            lsn < end_lsn
         });
 
         let mut frozen_layers: Vec<Arc<dyn Layer>> = Vec::new();