move the logic of emitting the clear visibility wal records into a common function

pageserver/src/walingest.rs

@@ -709,15 +709,34 @@ impl WalIngest {
             _ => {}
         }
 
-        // Clear the VM bits if required.
-
         let vm_rel = RelTag {
             forknum: VISIBILITYMAP_FORKNUM,
             spcnode: decoded.blocks[0].rnode_spcnode,
             dbnode: decoded.blocks[0].rnode_dbnode,
             relnode: decoded.blocks[0].rnode_relnode,
         };
+        self.clear_visibility_map_bits_if_required(
+            modification,
+            vm_rel,
+            new_heap_blkno,
+            old_heap_blkno,
+            flags,
+            ctx,
+        )
+        .await?;
 
+        Ok(())
+    }
+
+    async fn clear_visibility_map_bits_if_required(
+        &mut self,
+        modification: &mut DatadirModification<'_>,
+        vm_rel: RelTag,
+        new_heap_blkno: Option<u32>,
+        old_heap_blkno: Option<u32>,
+        flags: u8,
+        ctx: &RequestContext,
+    ) -> anyhow::Result<()> {
         let new = new_heap_blkno.map(|x| (x, pg_constants::HEAPBLK_TO_MAPBLOCK(x)));
         let old = old_heap_blkno.map(|x| (x, pg_constants::HEAPBLK_TO_MAPBLOCK(x)));
 
@@ -784,8 +803,7 @@ impl WalIngest {
                 }
             }
         }
-
-        Ok(())
+        anyhow::Ok(())
     }
 
     async fn ingest_neonrmgr_record(
@@ -874,81 +892,21 @@ impl WalIngest {
             ),
         }
 
-        // Clear the VM bits if required.
-
         let vm_rel = RelTag {
             forknum: VISIBILITYMAP_FORKNUM,
             spcnode: decoded.blocks[0].rnode_spcnode,
             dbnode: decoded.blocks[0].rnode_dbnode,
             relnode: decoded.blocks[0].rnode_relnode,
         };
-
-        let new = new_heap_blkno.map(|x| (x, pg_constants::HEAPBLK_TO_MAPBLOCK(x)));
-        let old = old_heap_blkno.map(|x| (x, pg_constants::HEAPBLK_TO_MAPBLOCK(x)));
-
-        // Sometimes, Postgres seems to create heap WAL records with the
-        // ALL_VISIBLE_CLEARED flag set, even though the bit in the VM page is
-        // not set. In fact, it's possible that the VM page does not exist at all.
-        // In that case, we don't want to store a record to clear the VM bit;
-        // replaying it would fail to find the previous image of the page, because
-        // it doesn't exist. So check if the VM page(s) exist, and skip the WAL
-        // record if it doesn't.
-        let (new, old) = {
-            let vm_size = if new.or(old).is_some() {
-                Some(get_relsize(modification, vm_rel, ctx).await?)
-            } else {
-                None
-            };
-            let filter = |(heap_blk, vm_blk)| {
-                let vm_size = vm_size.expect("we set it to Some() if new or old is Some()");
-                if vm_blk >= vm_size {
-                    None
-                } else {
-                    Some((heap_blk, vm_blk))
-                }
-            };
-            (new.and_then(filter), old.and_then(filter))
-        };
-
-        match (new, old) {
-            (Some((new_heap_blkno, new_vm_blk)), Some((old_heap_blkno, old_vm_blk)))
-                if new_vm_blk == old_vm_blk =>
-            {
-                // An UPDATE record that needs to clear the bits for both old and the
-                // new page, both of which reside on the same VM page.
-                self.put_rel_wal_record(
-                    modification,
-                    vm_rel,
-                    new_vm_blk, // could also be old_vm_blk, they're the same
-                    NeonWalRecord::ClearVisibilityMapFlags {
-                        heap_blkno_1: Some(new_heap_blkno),
-                        heap_blkno_2: Some(old_heap_blkno),
-                        flags,
-                    },
-                    ctx,
-                )
-                .await?;
-            }
-            (new, old) => {
-                // Emit one record per VM block that needs updating.
-                for update in [new, old] {
-                    if let Some((heap_blkno, vm_blk)) = update {
-                        self.put_rel_wal_record(
-                            modification,
-                            vm_rel,
-                            vm_blk,
-                            NeonWalRecord::ClearVisibilityMapFlags {
-                                heap_blkno_1: Some(heap_blkno),
-                                heap_blkno_2: None,
-                                flags,
-                            },
-                            ctx,
-                        )
-                        .await?;
-                    }
-                }
-            }
-        }
+        self.clear_visibility_map_bits_if_required(
+            modification,
+            vm_rel,
+            new_heap_blkno,
+            old_heap_blkno,
+            flags,
+            ctx,
+        )
+        .await?;
 
         Ok(())
     }