Issue #19. Bug in vacuuming.

- Incorrect adjusting of pd_lower in RumPageDeletePostingItem()
- Incorrect writing of Generic WAL record in rumDeletePage(). Because of 'goto restart' there was a memory
leak.

Author: za-arthur

src/rumdatapage.c

@@ -764,12 +764,14 @@ RumPageDeletePostingItem(Page page, OffsetNumber offset)
 				   *sourceptr = RumDataPageGetItem(page, offset + 1);
 
 		memmove(dstptr, sourceptr, sizeof(PostingItem) * (maxoff - offset));
-		/* Adjust pd_lower */
-		((PageHeader) page)->pd_lower = sourceptr - page;
-		Assert(((PageHeader) page)->pd_lower <= ((PageHeader) page)->pd_upper);
 	}
 
 	RumPageGetOpaque(page)->maxoff--;
+
+	/* Adjust pd_lower */
+	((PageHeader) page)->pd_lower =
+		RumDataPageGetItem(page, RumPageGetOpaque(page)->maxoff + 1) - page;
+	Assert(((PageHeader) page)->pd_lower <= ((PageHeader) page)->pd_upper);
 }
 
 /*
@@ -1274,7 +1276,7 @@ dataSplitPageInternal(RumBtree btree, Buffer lbuf, Buffer rbuf,
 	/* Adjust pd_lower */
 	((PageHeader) rPage)->pd_lower = (ptr +
 									  (maxoff - separator) * sizeofitem) -
-		rPage;
+									  rPage;
 
 	PostingItemSetBlockNumber(&(btree->pitem), BufferGetBlockNumber(lbuf));
 	if (RumPageIsLeaf(newlPage))

src/rumget.c

@@ -51,7 +51,7 @@ callAddInfoConsistentFn(RumState * rumstate, RumScanKey key)
 
 	key->outerAddInfoIsNull = true;
 
-	if (key->addInfoKeys == false && key->willSort == false)
+	if (key->addInfoKeys == NULL && key->willSort == false)
 		return true;
 
 	for (i = 0; i < key->nentries; i++)

src/rumvacuum.c

@@ -239,10 +239,6 @@ rumVacuumPostingTreeLeaves(RumVacuumState * gvs, OffsetNumber attnum,
 					oldMaxOff = RumPageGetOpaque(page)->maxoff;
 		Pointer		cleaned = NULL;
 		Size		newSize;
-		GenericXLogState *state;
-
-		state = GenericXLogStart(gvs->index);
-		page = GenericXLogRegisterBuffer(state, buffer, 0);
 
 		newMaxOff = rumVacuumPostingList(gvs, attnum,
 							   RumDataPageGetData(page), oldMaxOff, &cleaned,
@@ -251,21 +247,26 @@ rumVacuumPostingTreeLeaves(RumVacuumState * gvs, OffsetNumber attnum,
 		/* saves changes about deleted tuple ... */
 		if (oldMaxOff != newMaxOff)
 		{
+			GenericXLogState *state;
+			Page		newPage;
+
+			state = GenericXLogStart(gvs->index);
+
+			newPage = GenericXLogRegisterBuffer(state, buffer, 0);
+
 			if (newMaxOff > 0)
-				memcpy(RumDataPageGetData(page), cleaned, newSize);
+				memcpy(RumDataPageGetData(newPage), cleaned, newSize);
 
 			pfree(cleaned);
-			RumPageGetOpaque(page)->maxoff = newMaxOff;
-			updateItemIndexes(page, attnum, &gvs->rumstate);
+			RumPageGetOpaque(newPage)->maxoff = newMaxOff;
+			updateItemIndexes(newPage, attnum, &gvs->rumstate);
 
 			/* if root is a leaf page, we don't desire further processing */
-			if (!isRoot && RumPageGetOpaque(page)->maxoff < FirstOffsetNumber)
+			if (!isRoot && RumPageGetOpaque(newPage)->maxoff < FirstOffsetNumber)
 				hasVoidPage = TRUE;
 
 			GenericXLogFinish(state);
 		}
-		else
-			GenericXLogAbort(state);
 	}
 	else
 	{
@@ -341,8 +342,6 @@ rumDeletePage(RumVacuumState * gvs, BlockNumber deleteBlkno,
 
 	LockBuffer(dBuffer, RUM_UNLOCK);
 
-	state = GenericXLogStart(gvs->index);
-
 	/*
 	 * Lock the pages in the same order as an insertion would, to avoid
 	 * deadlocks: left, then right, then parent.
@@ -360,16 +359,16 @@ rumDeletePage(RumVacuumState * gvs, BlockNumber deleteBlkno,
 		UnlockReleaseBuffer(lBuffer);
 		ReleaseBuffer(dBuffer);
 		ReleaseBuffer(rBuffer);
+		ReleaseBuffer(pBuffer);
 		goto restart;
 	}
 	LockBuffer(rBuffer, RUM_EXCLUSIVE);
 	if (!isParentRoot)			/* parent is already locked by
 								 * LockBufferForCleanup() */
 		LockBuffer(pBuffer, RUM_EXCLUSIVE);
 
-	dPage = GenericXLogRegisterBuffer(state, dBuffer, 0);
-	lPage = GenericXLogRegisterBuffer(state, lBuffer, 0);
-	rPage = GenericXLogRegisterBuffer(state, rBuffer, 0);
+	lPage = BufferGetPage(lBuffer);
+	rPage = BufferGetPage(rBuffer);
 
 	/*
 	 * last chance to check
@@ -378,19 +377,29 @@ rumDeletePage(RumVacuumState * gvs, BlockNumber deleteBlkno,
 		  RumPageGetOpaque(rPage)->leftlink == deleteBlkno &&
 		  RumPageGetOpaque(dPage)->maxoff < FirstOffsetNumber))
 	{
+		OffsetNumber dMaxoff = RumPageGetOpaque(dPage)->maxoff;
+
 		if (!isParentRoot)
 			LockBuffer(pBuffer, RUM_UNLOCK);
 		ReleaseBuffer(pBuffer);
 		UnlockReleaseBuffer(lBuffer);
 		UnlockReleaseBuffer(dBuffer);
 		UnlockReleaseBuffer(rBuffer);
 
-		if (RumPageGetOpaque(dPage)->maxoff >= FirstOffsetNumber)
+		if (dMaxoff >= FirstOffsetNumber)
 			return false;
 
 		goto restart;
 	}
 
+	/* At least make the WAL record */
+
+	state = GenericXLogStart(gvs->index);
+
+	dPage = GenericXLogRegisterBuffer(state, dBuffer, 0);
+	lPage = GenericXLogRegisterBuffer(state, lBuffer, 0);
+	rPage = GenericXLogRegisterBuffer(state, rBuffer, 0);
+
 	RumPageGetOpaque(lPage)->rightlink = rightBlkno;
 	RumPageGetOpaque(rPage)->leftlink = leftBlkno;
 
@@ -484,9 +493,16 @@ rumScanToDelete(RumVacuumState * gvs, BlockNumber blkno, bool isRoot,
 	}
 
 	if (RumPageGetOpaque(page)->maxoff < FirstOffsetNumber && !isRoot)
+	{
+		/*
+		 * Release the buffer because in rumDeletePage() we need to pin it again
+		 * and call ConditionalLockBufferForCleanup().
+		 */
+		ReleaseBuffer(buffer);
 		meDelete = rumDeletePage(gvs, blkno, me->parent->blkno, myoff, me->parent->isRoot);
-
-	ReleaseBuffer(buffer);
+	}
+	else
+		ReleaseBuffer(buffer);
 
 	return meDelete;
 }