skip_vmbuffer_for_frozen_tuple_insertion_v3.patch

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 13396eb7f2..65ec6118b9 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2065,7 +2065,6 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 	Buffer		buffer;
 	Page		page = NULL;
 	Buffer		vmbuffer = InvalidBuffer;
-	bool		starting_with_empty_page;
 	bool		all_visible_cleared = false;
 	bool		all_frozen_set = false;
 	uint8		vmstatus = 0;
@@ -2082,8 +2081,8 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 	 * Find buffer to insert this tuple into.  If the page is all visible,
 	 * this will also pin the requisite visibility map page.
 	 *
-	 * Also pin visibility map page if COPY FREEZE inserts tuples into an
-	 * empty page. See all_frozen_set below.
+	 * Also pin visibility map page if we're inserting a frozen tuple into
+	 * an empty page. See all_frozen_set below.
 	 */
 	buffer = RelationGetBufferForTuple(relation, heaptup->t_len,
 									   InvalidBuffer, options, bistate,
@@ -2093,22 +2092,23 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 	/*
 	 * If we're inserting frozen entry into an empty page,
 	 * set visibility map bits and PageAllVisible() hint.
-	 *
-	 * If we're inserting frozen entry into already all_frozen page,
-	 * preserve this state.
 	 */
-	if (options & HEAP_INSERT_FROZEN)
+	if (options & HEAP_INSERT_FROZEN && BufferIsValid(vmbuffer))
 	{
-		page = BufferGetPage(buffer);
+		Assert(visibilitymap_pin_ok(BufferGetBlockNumber(buffer), vmbuffer));
 
-		starting_with_empty_page = PageGetMaxOffsetNumber(page) == 0;
+		page = BufferGetPage(buffer);
+		vmstatus = visibilitymap_get_status(relation,
+											BufferGetBlockNumber(buffer), &vmbuffer);
 
-		if (visibilitymap_pin_ok(BufferGetBlockNumber(buffer), vmbuffer))
-			vmstatus = visibilitymap_get_status(relation,
-								 BufferGetBlockNumber(buffer), &vmbuffer);
+		/* The page must be empty */
+		Assert(PageGetMaxOffsetNumber(page) == 0);
 
-		if ((starting_with_empty_page || vmstatus & VISIBILITYMAP_ALL_FROZEN))
-			all_frozen_set = true;
+		/*
+		 * If we're inserting frozen entry into empty page, we will set
+		 * all-visible to page and all-frozen on visibility map.
+		 */
+		all_frozen_set = true;
 	}
 
 	/*
diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index ffc89685bf..420503725c 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -301,6 +301,13 @@ RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
  *	Note that in some cases the caller might have already acquired such pins,
  *	which is indicated by these arguments not being InvalidBuffer on entry.
  *
+ *	In HEAP_INSERT_FROZEN cases, we handle the possibility that the caller will
+ *	sets all-frozen bit on the visibility map page.  We pin on the visibility
+ *	map page so that the caller can set the bit later, only if the page is
+ *	empty.  If the page is all-visible, we skip getting a pin on the
+ *	visibility map page while assuming the caller will neither clear nor set
+ *	the bit on the page.
+ *
  *	We normally use FSM to help us find free space.  However,
  *	if HEAP_INSERT_SKIP_FSM is specified, we just append a new empty page to
  *	the end of the relation if the tuple won't fit on the current target page.
@@ -425,6 +432,8 @@ RelationGetBufferForTuple(Relation relation, Size len,
 loop:
 	while (targetBlock != InvalidBlockNumber)
 	{
+		bool skip_vmbuffer = false;
+
 		/*
 		 * Read and exclusive-lock the target block, as well as the other
 		 * block if one was given, taking suitable care with lock ordering and
@@ -442,14 +451,28 @@ loop:
 		{
 			/* easy case */
 			buffer = ReadBufferBI(relation, targetBlock, RBM_NORMAL, bistate);
-			if (PageIsAllVisible(BufferGetPage(buffer)))
-				visibilitymap_pin(relation, targetBlock, vmbuffer);
 
-			/*
-			 * If the page is empty, pin vmbuffer to set all_frozen bit later.
-			 */
-			if ((options & HEAP_INSERT_FROZEN) &&
-				(PageGetMaxOffsetNumber(BufferGetPage(buffer)) == 0))
+			if (options & HEAP_INSERT_FROZEN)
+			{
+				/*
+				 * If we're inserting a frozen entry into empty page, pin on the
+				 * visibility map buffer to set all_frozen bit later.  Note that
+				 * we check without a buffer lock if the page is empty but the
+				 * caller doesn't need to recheck that since we assume that in
+				 * HEAP_INSERT_FROZEN case, only one process is inserting a
+				 * frozen tuple into this relation.
+				 *
+				 * If the page already is non-empty and all-visible, we skip to
+				 * pin on a visibility map buffer since we never clear and set
+				 * all-frozen bit on visibility map during inserting a frozen
+				 * tuple.
+				 */
+				if (PageGetMaxOffsetNumber(BufferGetPage(buffer)) == 0)
+					visibilitymap_pin(relation, targetBlock, vmbuffer);
+				else if (PageIsAllVisible(BufferGetPage(buffer)))
+					skip_vmbuffer = true;
+			}
+			else if (PageIsAllVisible(BufferGetPage(buffer)))
 				visibilitymap_pin(relation, targetBlock, vmbuffer);
 
 			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
@@ -502,7 +525,13 @@ loop:
 		 * done a bit of extra work for no gain, but there's no real harm
 		 * done.
 		 */
-		if (otherBuffer == InvalidBuffer || targetBlock <= otherBlock)
+		if (skip_vmbuffer)
+		{
+			/* We must be inserting a frozen tuple into the all-visible page */
+			Assert(options & HEAP_INSERT_FROZEN);
+			Assert(PageIsAllVisible(BufferGetPage(buffer)));
+		}
+		else if (otherBuffer == InvalidBuffer || targetBlock <= otherBlock)
 			GetVisibilityMapPins(relation, buffer, otherBuffer,
 								 targetBlock, otherBlock, vmbuffer,
 								 vmbuffer_other);
@@ -517,6 +546,18 @@ loop:
 		 */
 		page = BufferGetPage(buffer);
 
+#ifdef USE_ASSERT_CHECKING
+		/*
+		 * If we pin on the visibility map when inserting a frozen tuple,
+		 * the page must still be empty.
+		 */
+		if ((options & HEAP_INSERT_FROZEN) && BufferIsValid(*vmbuffer))
+		{
+			Assert(PageGetMaxOffsetNumber(page) == 0);
+			Assert(!BufferIsValid(*vmbuffer_other));
+		}
+#endif
+
 		/*
 		 * If necessary initialize page, it'll be used soon.  We could avoid
 		 * dirtying the buffer here, and rely on the caller to do so whenever