From 08978458beee31b23ffc7744fc2bdb60a692c9c0 Mon Sep 17 00:00:00 2001
From: Heikki Linnakangas <heikki@zenith.tech>
Date: Mon, 27 Sep 2021 19:15:44 +0300
Subject: [PATCH] Refactor write_to_disk, handling dropped segment as a special
 case.

Similar to what commit 7fb7f67b did to 'freeze', dealing with the
dropped segment separately from the rest of the logic makes the code
easier to follow. It is also needed by the next commit that replaces
the code to build new BTreeMap with an iterator; we cannot pass one
of two kinds of closures as argument, it has to always be the same one.
Having separate DeltaLayer::create() calls for the case of dropped
segment and the other cases works around that.
---
 .../src/layered_repository/inmemory_layer.rs  | 71 ++++++++++---------
 1 file changed, 39 insertions(+), 32 deletions(-)

diff --git a/pageserver/src/layered_repository/inmemory_layer.rs b/pageserver/src/layered_repository/inmemory_layer.rs
index c87ddcad0d..2499c9619b 100644
--- a/pageserver/src/layered_repository/inmemory_layer.rs
+++ b/pageserver/src/layered_repository/inmemory_layer.rs
@@ -668,37 +668,46 @@ impl InMemoryLayer {
         );
 
         let inner = self.inner.read().unwrap();
-
-        let drop_lsn = inner.drop_lsn;
-
         assert!(!inner.writeable);
 
-        let end_lsn = match drop_lsn {
-            Some(dlsn) => dlsn,
-            None => self.end_lsn.unwrap(),
-        };
-
         let predecessor = inner.predecessor.clone();
 
-        let mut before_page_versions;
-        let mut before_segsizes;
-        if inner.drop_lsn.is_none() {
-            before_segsizes = BTreeMap::new();
-            for (lsn, size) in inner.segsizes.iter() {
-                if *lsn <= end_lsn {
-                    before_segsizes.insert(*lsn, *size);
-                }
-            }
+        if let Some(drop_lsn) = inner.drop_lsn {
+            let delta_layer = DeltaLayer::create(
+                self.conf,
+                self.timelineid,
+                self.tenantid,
+                self.seg,
+                self.start_lsn,
+                drop_lsn,
+                true,
+                predecessor,
+                inner.page_versions.clone(),
+                inner.segsizes.clone(),
+            )?;
+            trace!(
+                "freeze: created delta layer for dropped segment {} {}-{}",
+                self.seg,
+                self.start_lsn,
+                drop_lsn
+            );
+            return Ok(vec![Arc::new(delta_layer)]);
+        }
 
-            before_page_versions = BTreeMap::new();
-            for ((blknum, lsn), pv) in inner.page_versions.iter() {
-                if *lsn < end_lsn {
-                    before_page_versions.insert((*blknum, *lsn), pv.clone());
-                }
+        let end_lsn = self.end_lsn.unwrap();
+
+        let mut before_segsizes = BTreeMap::new();
+        for (lsn, size) in inner.segsizes.iter() {
+            if *lsn <= end_lsn {
+                before_segsizes.insert(*lsn, *size);
+            }
+        }
+
+        let mut before_page_versions = BTreeMap::new();
+        for ((blknum, lsn), pv) in inner.page_versions.iter() {
+            if *lsn < end_lsn {
+                before_page_versions.insert((*blknum, *lsn), pv.clone());
             }
-        } else {
-            before_page_versions = inner.page_versions.clone();
-            before_segsizes = inner.segsizes.clone();
         }
 
         drop(inner);
@@ -714,7 +723,7 @@ impl InMemoryLayer {
                 self.seg,
                 self.start_lsn,
                 end_lsn,
-                drop_lsn.is_some(),
+                false,
                 predecessor,
                 before_page_versions,
                 before_segsizes,
@@ -730,12 +739,10 @@ impl InMemoryLayer {
             assert!(before_page_versions.is_empty());
         }
 
-        if drop_lsn.is_none() {
-            // Write a new base image layer at the cutoff point
-            let image_layer = ImageLayer::create_from_src(self.conf, timeline, self, end_lsn)?;
-            frozen_layers.push(Arc::new(image_layer));
-            trace!("freeze: created image layer {} at {}", self.seg, end_lsn);
-        }
+        // Write a new base image layer at the cutoff point
+        let image_layer = ImageLayer::create_from_src(self.conf, timeline, self, end_lsn)?;
+        frozen_layers.push(Arc::new(image_layer));
+        trace!("freeze: created image layer {} at {}", self.seg, end_lsn);
 
         Ok(frozen_layers)
     }