pageserver - serialize PageVersion as it is

Removes the need for PageVersionMeta struct.

pageserver/src/layered_repository/delta_layer.rs

@@ -42,12 +42,10 @@ use crate::layered_repository::filename::{DeltaFileName, PathOrConf};
 use crate::layered_repository::storage_layer::{
     Layer, PageReconstructData, PageReconstructResult, PageVersion, SegmentTag,
 };
-use crate::repository::WALRecord;
 use crate::waldecoder;
 use crate::PageServerConf;
 use crate::{ZTenantId, ZTimelineId};
 use anyhow::{bail, Result};
-use bytes::Bytes;
 use log::*;
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
@@ -109,12 +107,6 @@ impl From<&DeltaLayer> for Summary {
     }
 }
 
-#[derive(Serialize, Deserialize)]
-struct PageVersionMeta {
-    page_image_range: Option<BlobRange>,
-    record_range: Option<BlobRange>,
-}
-
 ///
 /// DeltaLayer is the in-memory data structure associated with an
 /// on-disk delta file.  We keep a DeltaLayer in memory for each
@@ -152,7 +144,7 @@ pub struct DeltaLayerInner {
 
     /// All versions of all pages in the file are are kept here.
     /// Indexed by block number and LSN.
-    page_version_metas: BTreeMap<(u32, Lsn), PageVersionMeta>,
+    page_version_metas: BTreeMap<(u32, Lsn), BlobRange>,
 
     /// `relsizes` tracks the size of the relation at different points in time.
     relsizes: BTreeMap<Lsn, u32>,
@@ -229,15 +221,15 @@ impl Layer for DeltaLayer {
             let mut iter = inner
                 .page_version_metas
                 .range((Included(&minkey), Included(&maxkey)));
-            while let Some(((_blknum, _entry_lsn), entry)) = iter.next_back() {
-                if let Some(img_range) = &entry.page_image_range {
+            while let Some(((_blknum, _entry_lsn), blob_range)) = iter.next_back() {
+                let pv = PageVersion::des(&read_blob(&page_version_reader, blob_range)?)?;
+
+                if let Some(img) = pv.page_image {
                     // Found a page image, return it
-                    let img = Bytes::from(read_blob(&page_version_reader, img_range)?);
                     reconstruct_data.page_img = Some(img);
                     need_image = false;
                     break;
-                } else if let Some(rec_range) = &entry.record_range {
-                    let rec = WALRecord::des(&read_blob(&page_version_reader, rec_range)?)?;
+                } else if let Some(rec) = pv.record {
                     let will_init = rec.will_init;
                     reconstruct_data.records.push(rec);
                     if will_init {
@@ -340,16 +332,16 @@ impl Layer for DeltaLayer {
         println!("--- page versions ---");
         let (_path, book) = self.open_book()?;
         let chapter = book.chapter_reader(PAGE_VERSIONS_CHAPTER)?;
-        for (k, v) in inner.page_version_metas.iter() {
+        for ((blk, lsn), blob_range) in inner.page_version_metas.iter() {
             let mut desc = String::new();
 
-            if let Some(page_image_range) = v.page_image_range.as_ref() {
-                let image = read_blob(&chapter, page_image_range)?;
-                write!(&mut desc, " img {} bytes", image.len())?;
+            let buf = read_blob(&chapter, blob_range)?;
+            let pv = PageVersion::des(&buf)?;
+
+            if let Some(img) = pv.page_image.as_ref() {
+                write!(&mut desc, " img {} bytes", img.len())?;
             }
-            if let Some(record_range) = v.record_range.as_ref() {
-                let record_bytes = read_blob(&chapter, record_range)?;
-                let rec = WALRecord::des(&record_bytes)?;
+            if let Some(rec) = pv.record.as_ref() {
                 let wal_desc = waldecoder::describe_wal_record(&rec.rec);
                 write!(
                     &mut desc,
@@ -359,7 +351,7 @@ impl Layer for DeltaLayer {
                     wal_desc
                 )?;
             }
-            println!("  blk {} at {}: {}", k.0, k.1, desc);
+            println!("  blk {} at {}: {}", blk, lsn, desc);
         }
 
         Ok(())
@@ -432,28 +424,10 @@ impl DeltaLayer {
         let mut page_version_writer = BlobWriter::new(book, PAGE_VERSIONS_CHAPTER);
 
         for (key, page_version) in page_versions {
-            let page_image_range = page_version
-                .page_image
-                .as_ref()
-                .map(|page_image| page_version_writer.write_blob(page_image))
-                .transpose()?;
+            let buf = PageVersion::ser(page_version)?;
+            let blob_range = page_version_writer.write_blob(&buf)?;
 
-            let record_range = page_version
-                .record
-                .as_ref()
-                .map(|record| {
-                    let buf = WALRecord::ser(record)?;
-                    page_version_writer.write_blob(&buf)
-                })
-                .transpose()?;
-
-            let old = inner.page_version_metas.insert(
-                *key,
-                PageVersionMeta {
-                    page_image_range,
-                    record_range,
-                },
-            );
+            let old = inner.page_version_metas.insert(*key, blob_range);
 
             assert!(old.is_none());
         }