Fix Generic WAL for vacuum

Artur Zakirov · Artur Zakirov · commit aaa4b0240112 · 2016-04-22T15:49:20.000+03:00
diff --git a/rumfast.c b/rumfast.c
@@ -528,7 +528,8 @@ shiftList(Relation index, Buffer metabuffer, BlockNumber newHead,
 			buffers[data.ndeleted] = ReadBuffer(index, blknoToDelete);
 			LockBuffer(buffers[data.ndeleted], RUM_EXCLUSIVE);
 
-			page = BufferGetPage(buffers[data.ndeleted]);
+			page = BufferGetPage(buffers[data.ndeleted], NULL, NULL,
+								 BGP_NO_SNAPSHOT_TEST);
 
 			data.ndeleted++;
 
@@ -739,7 +740,7 @@ rumInsertCleanup(RumState *rumstate,
 	metabuffer = ReadBuffer(index, RUM_METAPAGE_BLKNO);
 	LockBuffer(metabuffer, RUM_SHARE);
 
-	metapage = BufferGetPage(metabuffer);
+	metapage = BufferGetPage(metabuffer, NULL, NULL, BGP_NO_SNAPSHOT_TEST);
 	metadata = RumPageGetMeta(metapage);
 
 	if (metadata->head == InvalidBlockNumber)
@@ -755,7 +756,7 @@ rumInsertCleanup(RumState *rumstate,
 	blkno = metadata->head;
 	buffer = ReadBuffer(index, blkno);
 	LockBuffer(buffer, RUM_SHARE);
-	page = BufferGetPage(buffer);
+	page = BufferGetPage(buffer, NULL, NULL, BGP_NO_SNAPSHOT_TEST);
 
 	LockBuffer(metabuffer, RUM_UNLOCK);
 
@@ -941,7 +942,7 @@ rumInsertCleanup(RumState *rumstate,
 		vacuum_delay_point();
 		buffer = ReadBuffer(index, blkno);
 		LockBuffer(buffer, RUM_SHARE);
-		page = BufferGetPage(buffer);
+		page = BufferGetPage(buffer, NULL, NULL, BGP_NO_SNAPSHOT_TEST);
 	}
 
 	ReleaseBuffer(metabuffer);
diff --git a/rumvacuum.c b/rumvacuum.c
@@ -274,8 +274,12 @@ rumVacuumPostingTreeLeaves(RumVacuumState *gvs, OffsetNumber attnum, BlockNumber
 	{
 		OffsetNumber newMaxOff,
 					oldMaxOff = RumPageGetOpaque(page)->maxoff;
-		Pointer cleaned = NULL;
-		Size newSize;
+		Pointer		cleaned = NULL;
+		Size		newSize;
+		GenericXLogState *state;
+
+		state = GenericXLogStart(gvs->index);
+		page = GenericXLogRegisterBuffer(state, buffer, 0);
 
 		newMaxOff = rumVacuumPostingList(gvs, attnum,
 				RumDataPageGetData(page), oldMaxOff, &cleaned,
@@ -284,24 +288,21 @@ rumVacuumPostingTreeLeaves(RumVacuumState *gvs, OffsetNumber attnum, BlockNumber
 		/* saves changes about deleted tuple ... */
 		if (oldMaxOff != newMaxOff)
 		{
-			START_CRIT_SECTION();
-
 			if (newMaxOff > 0)
 				memcpy(RumDataPageGetData(page), cleaned, newSize);
 
 			pfree(cleaned);
 			RumPageGetOpaque(page)->maxoff = newMaxOff;
 			updateItemIndexes(page, attnum, &gvs->rumstate);
 
-			MarkBufferDirty(buffer);
-			xlogVacuumPage(gvs->index, buffer, attnum, &gvs->rumstate);
-
-			END_CRIT_SECTION();
+			GenericXLogFinish(state);
 
 			/* if root is a leaf page, we don't desire further processing */
 			if (!isRoot && RumPageGetOpaque(page)->maxoff < FirstOffsetNumber)
 				hasVoidPage = TRUE;
 		}
+		else
+			GenericXLogAbort(state);
 	}
 	else
 	{
@@ -691,11 +692,14 @@ rumbulkdelete(IndexVacuumInfo *info,
 
 	for (;;)
 	{
-		Page		page = BufferGetPage(buffer, NULL, NULL,
-										 BGP_NO_SNAPSHOT_TEST);
+		GenericXLogState *state;
+		Page		page;
 		Page		resPage;
 		uint32		i;
 
+		state = GenericXLogStart(index);
+		page = GenericXLogRegisterBuffer(state, buffer, 0);
+
 		Assert(!RumPageIsData(page));
 
 		resPage = rumVacuumEntryPage(&gvs, buffer, rootOfPostingTree, attnumOfPostingTree, &nRoot);
@@ -704,15 +708,13 @@ rumbulkdelete(IndexVacuumInfo *info,
 
 		if (resPage)
 		{
-			START_CRIT_SECTION();
 			PageRestoreTempPage(resPage, page);
-			MarkBufferDirty(buffer);
-			xlogVacuumPage(gvs.index, buffer, InvalidOffsetNumber, &gvs.rumstate);
+			GenericXLogFinish(state);
 			UnlockReleaseBuffer(buffer);
-			END_CRIT_SECTION();
 		}
 		else
 		{
+			GenericXLogAbort(state);
 			UnlockReleaseBuffer(buffer);
 		}
 
