rumbuild, rumbuildempty, ruminsert work. Need to check rumInsertCleanup, vacuum

Artur Zakirov · Artur Zakirov · commit 14a6bce84102 · 2016-04-21T21:12:27.000+03:00
diff --git a/rum.h b/rum.h
@@ -14,6 +14,7 @@
 
 #include "access/amapi.h"
 #include "access/genam.h"
+#include "access/generic_xlog.h"
 #include "access/gin.h"
 #include "access/itup.h"
 #include "fmgr.h"
@@ -506,9 +507,9 @@ extern bytea *rumoptions(Datum reloptions, bool validate);
 extern Datum rumhandler(PG_FUNCTION_ARGS);
 extern void initRumState(RumState *state, Relation index);
 extern Buffer RumNewBuffer(Relation index);
-extern void RumInitBuffer(Buffer buffer, uint32 flags);
+extern void RumInitBuffer(GenericXLogState *state, Buffer buffer, uint32 flags);
 extern void RumInitPage(Page page, uint32 f, Size pageSize);
-extern void RumInitMetabuffer(Relation index, Buffer metaBuffer);
+extern void RumInitMetabuffer(GenericXLogState *state, Buffer metaBuffer);
 extern int rumCompareEntries(RumState *rumstate, OffsetNumber attnum,
 				  Datum a, RumNullCategory categorya,
 				  Datum b, RumNullCategory categoryb);
diff --git a/rumbtree.c b/rumbtree.c
@@ -431,7 +431,6 @@ rumInsertValue(Relation index, RumBtree btree, RumBtreeStack *stack,
 
 				rpage = GenericXLogRegisterBuffer(state, rbuffer, 0);
 
-
 				RumPageGetOpaque(rpage)->rightlink = InvalidBlockNumber;
 				RumPageGetOpaque(newlpage)->rightlink = BufferGetBlockNumber(rbuffer);
 				((rumxlogSplit *) (rdata->data))->lblkno = BufferGetBlockNumber(lbuffer);
diff --git a/rumdatapage.c b/rumdatapage.c
@@ -592,6 +592,7 @@ RumDataPageAddItem(Page page, void *data, OffsetNumber offset)
 					(maxoff - offset + 1) * RumSizeOfDataPageItem(page));
 	}
 	memcpy(ptr, data, RumSizeOfDataPageItem(page));
+	((PageHeader) page)->pd_lower = ptr - page;
 
 	RumPageGetOpaque(page)->maxoff++;
 }
diff --git a/rumfast.c b/rumfast.c
@@ -46,39 +46,27 @@ typedef struct KeyArray
  * Returns amount of free space left on the page.
  */
 static int32
-writeListPage(Relation index, Buffer buffer, GenericXLogState *state,
+writeListPage(Relation index, Buffer buffer,
 			  IndexTuple *tuples, int32 ntuples, BlockNumber rightlink)
 {
 	Page		page;
 	int32		i,
-				freesize,
-				size = 0;
-	OffsetNumber l,
-				off;
-	char	   *workspace;
-	char	   *ptr;
-	GenericXLogState *state2;
+				freesize;
+	OffsetNumber		l,
+						off;
+	GenericXLogState   *state;
 
-	state2 = GenericXLogStart(index);
-	/* workspace could be a local array; we use palloc for alignment */
-	workspace = palloc(BLCKSZ);
+	state = GenericXLogStart(index);
 
-	START_CRIT_SECTION();
-
-	RumInitBuffer(buffer, RUM_LIST);
-	page = GenericXLogRegisterBuffer(state2, buffer, 0);
+	page = GenericXLogRegisterBuffer(state, buffer, 0);
+	RumInitPage(page, RUM_LIST, BufferGetPageSize(buffer));
 
 	off = FirstOffsetNumber;
-	ptr = workspace;
 
 	for (i = 0; i < ntuples; i++)
 	{
 		int			this_size = IndexTupleSize(tuples[i]);
 
-		memcpy(ptr, tuples[i], this_size);
-		ptr += this_size;
-		size += this_size;
-
 		l = PageAddItem(page, (Item) tuples[i], this_size, off, false, false);
 
 		if (l == InvalidOffsetNumber)
@@ -88,8 +76,6 @@ writeListPage(Relation index, Buffer buffer, GenericXLogState *state,
 		off++;
 	}
 
-	Assert(size <= BLCKSZ);		/* else we overran workspace */
-
 	RumPageGetOpaque(page)->rightlink = rightlink;
 
 	/*
@@ -107,24 +93,17 @@ writeListPage(Relation index, Buffer buffer, GenericXLogState *state,
 		RumPageGetOpaque(page)->maxoff = 0;
 	}
 
-// 	MarkBufferDirty(buffer);
-	GenericXLogFinish(state2);
-
 	/* get free space before releasing buffer */
 	freesize = PageGetExactFreeSpace(page);
-
+	GenericXLogFinish(state);
 	UnlockReleaseBuffer(buffer);
 
-	END_CRIT_SECTION();
-
-	pfree(workspace);
-
 	return freesize;
 }
 
 static void
-makeSublist(Relation index, GenericXLogState *state,
-			IndexTuple *tuples, int32 ntuples, RumMetaPageData *res)
+makeSublist(Relation index, IndexTuple *tuples, int32 ntuples,
+			RumMetaPageData *res)
 {
 	Buffer		curBuffer = InvalidBuffer;
 	Buffer		prevBuffer = InvalidBuffer;
@@ -147,7 +126,7 @@ makeSublist(Relation index, GenericXLogState *state,
 			if (prevBuffer != InvalidBuffer)
 			{
 				res->nPendingPages++;
-				writeListPage(index, prevBuffer, state,
+				writeListPage(index, prevBuffer,
 							  tuples + startTuple,
 							  i - startTuple,
 							  BufferGetBlockNumber(curBuffer));
@@ -180,7 +159,7 @@ makeSublist(Relation index, GenericXLogState *state,
 	 * Write last page
 	 */
 	res->tail = BufferGetBlockNumber(curBuffer);
-	res->tailFreeSize = writeListPage(index, curBuffer, state,
+	res->tailFreeSize = writeListPage(index, curBuffer,
 									  tuples + startTuple,
 									  ntuples - startTuple,
 									  InvalidBlockNumber);
@@ -203,29 +182,17 @@ rumHeapTupleFastInsert(RumState *rumstate, RumTupleCollector *collector)
 	Buffer		metabuffer;
 	Page		metapage;
 	RumMetaPageData *metadata = NULL;
-// 	RumXLogRecData rdata[2];
 	Buffer		buffer = InvalidBuffer;
 	Page		page = NULL;
-	rumxlogUpdateMeta data;
 	bool		separateList = false;
 	bool		needCleanup = false;
 	GenericXLogState *state;
 
 	if (collector->ntuples == 0)
 		return;
 
-	data.node = index->rd_node;
-	data.ntuples = 0;
-	data.newRightlink = data.prevTail = InvalidBlockNumber;
-
-// 	rdata[0].buffer = InvalidBuffer;
-// 	rdata[0].data = (char *) &data;
-// 	rdata[0].len = sizeof(rumxlogUpdateMeta);
-// 	rdata[0].next = NULL;
-
 	state = GenericXLogStart(rumstate->index);
 	metabuffer = ReadBuffer(index, RUM_METAPAGE_BLKNO);
-	metapage = GenericXLogRegisterBuffer(state, metabuffer, 0);
 
 	if (collector->sumsize + collector->ntuples * sizeof(ItemIdData) > RumListPageSize)
 	{
@@ -237,7 +204,8 @@ rumHeapTupleFastInsert(RumState *rumstate, RumTupleCollector *collector)
 	else
 	{
 		LockBuffer(metabuffer, RUM_EXCLUSIVE);
-		metadata = RumPageGetMeta(metapage);
+		metadata = RumPageGetMeta(BufferGetPage(metabuffer, NULL, NULL,
+								  BGP_NO_SNAPSHOT_TEST));
 
 		if (metadata->head == InvalidBlockNumber ||
 			collector->sumsize + collector->ntuples * sizeof(ItemIdData) > metadata->tailFreeSize)
@@ -261,22 +229,20 @@ rumHeapTupleFastInsert(RumState *rumstate, RumTupleCollector *collector)
 		RumMetaPageData sublist;
 
 		memset(&sublist, 0, sizeof(RumMetaPageData));
-		makeSublist(index, state,
-					collector->tuples, collector->ntuples, &sublist);
+		makeSublist(index, collector->tuples, collector->ntuples, &sublist);
 
 		/*
 		 * metapage was unlocked, see above
 		 */
 		LockBuffer(metabuffer, RUM_EXCLUSIVE);
+		metapage = GenericXLogRegisterBuffer(state, metabuffer, 0);
 		metadata = RumPageGetMeta(metapage);
 
 		if (metadata->head == InvalidBlockNumber)
 		{
 			/*
 			 * Main list is empty, so just insert sublist as main list
 			 */
-			START_CRIT_SECTION();
-
 			metadata->head = sublist.head;
 			metadata->tail = sublist.tail;
 			metadata->tailFreeSize = sublist.tailFreeSize;
@@ -289,29 +255,15 @@ rumHeapTupleFastInsert(RumState *rumstate, RumTupleCollector *collector)
 			/*
 			 * Merge lists
 			 */
-			data.prevTail = metadata->tail;
-			data.newRightlink = sublist.head;
 
 			buffer = ReadBuffer(index, metadata->tail);
 			LockBuffer(buffer, RUM_EXCLUSIVE);
 			page = GenericXLogRegisterBuffer(state, buffer, 0);
 
-// 			rdata[0].next = rdata + 1;
-//
-// 			rdata[1].buffer = buffer;
-// 			rdata[1].buffer_std = true;
-// 			rdata[1].data = NULL;
-// 			rdata[1].len = 0;
-// 			rdata[1].next = NULL;
-
 			Assert(RumPageGetOpaque(page)->rightlink == InvalidBlockNumber);
 
-			START_CRIT_SECTION();
-
 			RumPageGetOpaque(page)->rightlink = sublist.head;
 
-// 			MarkBufferDirty(buffer);
-
 			metadata->tail = sublist.tail;
 			metadata->tailFreeSize = sublist.tailFreeSize;
 
@@ -328,7 +280,9 @@ rumHeapTupleFastInsert(RumState *rumstate, RumTupleCollector *collector)
 					off;
 		int			i,
 					tupsize;
-		char	   *ptr;
+
+		metapage = GenericXLogRegisterBuffer(state, metabuffer, 0);
+		metadata = RumPageGetMeta(metapage);
 
 		buffer = ReadBuffer(index, metadata->tail);
 		LockBuffer(buffer, RUM_EXCLUSIVE);
@@ -337,18 +291,6 @@ rumHeapTupleFastInsert(RumState *rumstate, RumTupleCollector *collector)
 		off = (PageIsEmpty(page)) ? FirstOffsetNumber :
 			OffsetNumberNext(PageGetMaxOffsetNumber(page));
 
-// 		rdata[0].next = rdata + 1;
-//
-// 		rdata[1].buffer = buffer;
-// 		rdata[1].buffer_std = true;
-		ptr /*= rdata[1].data*/ = (char *) palloc(collector->sumsize);
-// 		rdata[1].len = collector->sumsize;
-// 		rdata[1].next = NULL;
-
-		data.ntuples = collector->ntuples;
-
-		START_CRIT_SECTION();
-
 		/*
 		 * Increase counter of heap tuples
 		 */
@@ -368,29 +310,12 @@ rumHeapTupleFastInsert(RumState *rumstate, RumTupleCollector *collector)
 					 RelationGetRelationName(index));
 			}
 
-			memcpy(ptr, collector->tuples[i], tupsize);
-			ptr += tupsize;
-
 			off++;
 		}
 
-// 		Assert((ptr - rdata[1].data) <= collector->sumsize);
-
 		metadata->tailFreeSize = PageGetExactFreeSpace(page);
-
-// 		MarkBufferDirty(buffer);
 	}
 
-	/*
-	 * Write metabuffer, make xlog entry
-	 */
-// 	MarkBufferDirty(metabuffer);
-
-	GenericXLogFinish(state);
-
-	if (buffer != InvalidBuffer)
-		UnlockReleaseBuffer(buffer);
-
 	/*
 	 * Force pending list cleanup when it becomes too long. And,
 	 * rumInsertCleanup could take significant amount of time, so we prefer to
@@ -403,9 +328,12 @@ rumHeapTupleFastInsert(RumState *rumstate, RumTupleCollector *collector)
 	if (metadata->nPendingPages * RUM_PAGE_FREESIZE > work_mem * 1024L)
 		needCleanup = true;
 
-	UnlockReleaseBuffer(metabuffer);
+	GenericXLogFinish(state);
 
-	END_CRIT_SECTION();
+	if (buffer != InvalidBuffer)
+		UnlockReleaseBuffer(buffer);
+
+	UnlockReleaseBuffer(metabuffer);
 
 	if (needCleanup)
 		rumInsertCleanup(rumstate, false, NULL);
diff --git a/ruminsert.c b/ruminsert.c
diff --git a/rumutil.c b/rumutil.c

Original file line number	Diff line number	Diff line change
`@@ -592,6 +592,7 @@ RumDataPageAddItem(Page page, void *data, OffsetNumber offset)`
`592`	`592`	`(maxoff - offset + 1) * RumSizeOfDataPageItem(page));`
`593`	`593`	`}`
`594`	`594`	`memcpy(ptr, data, RumSizeOfDataPageItem(page));`
	`595`	`+ ((PageHeader) page)->pd_lower = ptr - page;`
`595`	`596`
`596`	`597`	`RumPageGetOpaque(page)->maxoff++;`
`597`	`598`	`}`