From 70a1d6888fb320ba210f76d31bf1736c7182b1d3 Mon Sep 17 00:00:00 2001
From: David Christensen <david.christensen@crunchydata.com>
Date: Fri, 29 Sep 2023 15:03:00 -0400
Subject: [PATCH v4 17/22] optimization: Use fastdiv code in visibility map

Adjust the code that calculates our heap block offsets based to be based on
PageUsableSpace instead of compile-time constants.  Use the fastdiv code to
support this.
---
 src/backend/access/heap/visibilitymap.c | 95 +++++++++++++++++--------
 1 file changed, 67 insertions(+), 28 deletions(-)

diff --git a/src/backend/access/heap/visibilitymap.c b/src/backend/access/heap/visibilitymap.c
index 6bde6f4388..8c47076548 100644
--- a/src/backend/access/heap/visibilitymap.c
+++ b/src/backend/access/heap/visibilitymap.c
@@ -105,18 +105,19 @@
  * extra headers, so the whole page minus the standard page header is
  * used for the bitmap.
  */
-#define MAPSIZE (PageUsableSpace)
 
-/* Number of heap blocks we can represent in one byte */
-#define HEAPBLOCKS_PER_BYTE (BITS_PER_BYTE / BITS_PER_HEAPBLOCK)
-
-/* Number of heap blocks we can represent in one visibility map page. */
-#define HEAPBLOCKS_PER_PAGE (MAPSIZE * HEAPBLOCKS_PER_BYTE)
+#define MAPBLOCK_SIZE PageUsableSpace
+#define HEAPBLOCK_SIZE (MAPBLOCK_SIZE<<2)
+/* Init routine for our fastmath */
+#define MAPBLOCK_INIT if (unlikely(!heapblock_inv))		\
+	{													\
+		heapblock_inv = pg_fastinverse(HEAPBLOCK_SIZE);	\
+	}
 
 /* Mapping from heap block number to the right bit in the visibility map */
-#define HEAPBLK_TO_MAPBLOCK(x) ((x) / HEAPBLOCKS_PER_PAGE)
-#define HEAPBLK_TO_MAPBYTE(x) (((x) % HEAPBLOCKS_PER_PAGE) / HEAPBLOCKS_PER_BYTE)
-#define HEAPBLK_TO_OFFSET(x) (((x) % HEAPBLOCKS_PER_BYTE) * BITS_PER_HEAPBLOCK)
+#define HEAPBLK_TO_MAPBLOCK(x) (pg_fastdiv((x),HEAPBLOCK_SIZE,heapblock_inv))
+#define HEAPBLK_TO_MAPBYTE(x) (pg_fastmod((x),HEAPBLOCK_SIZE,heapblock_inv) >> 2) /* always 4 blocks per byte */
+#define HEAPBLK_TO_OFFSET(x) (((x) & 0x3) << 1) /* always 2 bits per entry */
 
 /* Masks for counting subsets of bits in the visibility map. */
 #define VISIBLE_MASK64	UINT64CONST(0x5555555555555555) /* The lower bit of each
@@ -128,6 +129,8 @@
 static Buffer vm_readbuf(Relation rel, BlockNumber blkno, bool extend);
 static Buffer vm_extend(Relation rel, BlockNumber vm_nblocks);
 
+/* storage for the fast div/mod inverse */
+static uint64 heapblock_inv = 0;
 
 /*
  *	visibilitymap_clear - clear specified bits for one page in visibility map
@@ -139,13 +142,20 @@ static Buffer vm_extend(Relation rel, BlockNumber vm_nblocks);
 bool
 visibilitymap_clear(Relation rel, BlockNumber heapBlk, Buffer vmbuf, uint8 flags)
 {
-	BlockNumber mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
-	int			mapByte = HEAPBLK_TO_MAPBYTE(heapBlk);
-	int			mapOffset = HEAPBLK_TO_OFFSET(heapBlk);
-	uint8		mask = flags << mapOffset;
+	BlockNumber mapBlock;
+	int			mapByte;
+	int			mapOffset;
+	uint8		mask;
 	char	   *map;
 	bool		cleared = false;
 
+	MAPBLOCK_INIT;
+
+	mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
+	mapByte = HEAPBLK_TO_MAPBYTE(heapBlk);
+	mapOffset = HEAPBLK_TO_OFFSET(heapBlk);
+	mask = flags << mapOffset;
+
 	/* Must never clear all_visible bit while leaving all_frozen bit set */
 	Assert(flags & VISIBILITYMAP_VALID_BITS);
 	Assert(flags != VISIBILITYMAP_ALL_VISIBLE);
@@ -192,7 +202,11 @@ visibilitymap_clear(Relation rel, BlockNumber heapBlk, Buffer vmbuf, uint8 flags
 void
 visibilitymap_pin(Relation rel, BlockNumber heapBlk, Buffer *vmbuf)
 {
-	BlockNumber mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
+	BlockNumber mapBlock;
+
+	MAPBLOCK_INIT;
+
+	mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
 
 	/* Reuse the old pinned buffer if possible */
 	if (BufferIsValid(*vmbuf))
@@ -216,7 +230,11 @@ visibilitymap_pin(Relation rel, BlockNumber heapBlk, Buffer *vmbuf)
 bool
 visibilitymap_pin_ok(BlockNumber heapBlk, Buffer vmbuf)
 {
-	BlockNumber mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
+	BlockNumber mapBlock;
+
+	MAPBLOCK_INIT;
+
+	mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
 
 	return BufferIsValid(vmbuf) && BufferGetBlockNumber(vmbuf) == mapBlock;
 }
@@ -247,12 +265,18 @@ visibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf,
 				  XLogRecPtr recptr, Buffer vmBuf, TransactionId cutoff_xid,
 				  uint8 flags)
 {
-	BlockNumber mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
-	uint32		mapByte = HEAPBLK_TO_MAPBYTE(heapBlk);
-	uint8		mapOffset = HEAPBLK_TO_OFFSET(heapBlk);
+	BlockNumber mapBlock;
+	uint32		mapByte;
+	uint8		mapOffset;
 	Page		page;
 	uint8	   *map;
 
+	MAPBLOCK_INIT;
+
+	mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
+	mapByte = HEAPBLK_TO_MAPBYTE(heapBlk);
+	mapOffset = HEAPBLK_TO_OFFSET(heapBlk);
+
 #ifdef TRACE_VISIBILITYMAP
 	elog(DEBUG1, "vm_set %s %d", RelationGetRelationName(rel), heapBlk);
 #endif
@@ -337,12 +361,18 @@ visibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf,
 uint8
 visibilitymap_get_status(Relation rel, BlockNumber heapBlk, Buffer *vmbuf)
 {
-	BlockNumber mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
-	uint32		mapByte = HEAPBLK_TO_MAPBYTE(heapBlk);
-	uint8		mapOffset = HEAPBLK_TO_OFFSET(heapBlk);
+	BlockNumber mapBlock;
+	uint32		mapByte;
+	uint8		mapOffset;
 	char	   *map;
 	uint8		result;
 
+	MAPBLOCK_INIT;
+
+	mapBlock = HEAPBLK_TO_MAPBLOCK(heapBlk);
+	mapByte = HEAPBLK_TO_MAPBYTE(heapBlk);
+	mapOffset = HEAPBLK_TO_OFFSET(heapBlk);
+
 #ifdef TRACE_VISIBILITYMAP
 	elog(DEBUG1, "vm_get_status %s %d", RelationGetRelationName(rel), heapBlk);
 #endif
@@ -389,6 +419,8 @@ visibilitymap_count(Relation rel, BlockNumber *all_visible, BlockNumber *all_fro
 	BlockNumber nvisible = 0;
 	BlockNumber nfrozen = 0;
 
+	MAPBLOCK_INIT;
+
 	/* all_visible must be specified */
 	Assert(all_visible);
 
@@ -414,15 +446,16 @@ visibilitymap_count(Relation rel, BlockNumber *all_visible, BlockNumber *all_fro
 		 */
 		map = (uint64 *) PageGetContents(BufferGetPage(mapBuffer));
 
-		Assert(MAPSIZE % sizeof(uint64) == 0);
+		Assert(MAPBLOCK_SIZE % sizeof(uint64) == 0);
+
 		if (all_frozen == NULL)
 		{
-			for (i = 0; i < MAPSIZE / sizeof(uint64); i++)
+			for (i = 0; i < MAPBLOCK_SIZE / sizeof(uint64); i++)
 				nvisible += pg_popcount64(map[i] & VISIBLE_MASK64);
 		}
 		else
 		{
-			for (i = 0; i < MAPSIZE / sizeof(uint64); i++)
+			for (i = 0; i < MAPBLOCK_SIZE / sizeof(uint64); i++)
 			{
 				nvisible += pg_popcount64(map[i] & VISIBLE_MASK64);
 				nfrozen += pg_popcount64(map[i] & FROZEN_MASK64);
@@ -454,9 +487,15 @@ visibilitymap_prepare_truncate(Relation rel, BlockNumber nheapblocks)
 	BlockNumber newnblocks;
 
 	/* last remaining block, byte, and bit */
-	BlockNumber truncBlock = HEAPBLK_TO_MAPBLOCK(nheapblocks);
-	uint32		truncByte = HEAPBLK_TO_MAPBYTE(nheapblocks);
-	uint8		truncOffset = HEAPBLK_TO_OFFSET(nheapblocks);
+	BlockNumber truncBlock;
+	uint32		truncByte;
+	uint8		truncOffset;
+
+	MAPBLOCK_INIT;
+
+	truncBlock = HEAPBLK_TO_MAPBLOCK(nheapblocks);
+	truncByte = HEAPBLK_TO_MAPBYTE(nheapblocks);
+	truncOffset = HEAPBLK_TO_OFFSET(nheapblocks);
 
 #ifdef TRACE_VISIBILITYMAP
 	elog(DEBUG1, "vm_truncate %s %d", RelationGetRelationName(rel), nheapblocks);
@@ -500,7 +539,7 @@ visibilitymap_prepare_truncate(Relation rel, BlockNumber nheapblocks)
 		START_CRIT_SECTION();
 
 		/* Clear out the unwanted bytes. */
-		MemSet(&map[truncByte + 1], 0, MAPSIZE - (truncByte + 1));
+		MemSet(&map[truncByte + 1], 0, MAPBLOCK_SIZE - (truncByte + 1));
 
 		/*----
 		 * Mask out the unwanted bits of the last remaining byte.
-- 
2.40.1

