diff -c -r -N src.orig/backend/access/common/reloptions.c src/backend/access/common/reloptions.c
*** src.orig/backend/access/common/reloptions.c	2009-02-04 18:13:28.000000000 +0300
--- src/backend/access/common/reloptions.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 48,53 ****
--- 48,61 ----
  
  static relopt_bool boolRelOpts[] =
  {
+ 	{
+ 		{
+ 			"fastupdate",
+ 			"Enable/disable fast update mode of GIN index",
+ 			RELOPT_KIND_GIN
+ 		},
+ 		true
+ 	},
  	/* list terminator */
  	{ { NULL } }
  };
diff -c -r -N src.orig/backend/access/gin/Makefile src/backend/access/gin/Makefile
*** src.orig/backend/access/gin/Makefile	2009-02-04 18:13:28.000000000 +0300
--- src/backend/access/gin/Makefile	2009-02-04 18:23:22.000000000 +0300
***************
*** 14,19 ****
  
  OBJS = ginutil.o gininsert.o ginxlog.o ginentrypage.o gindatapage.o \
  	ginbtree.o ginscan.o ginget.o ginvacuum.o ginarrayproc.o \
! 	ginbulk.o
  
  include $(top_srcdir)/src/backend/common.mk
--- 14,19 ----
  
  OBJS = ginutil.o gininsert.o ginxlog.o ginentrypage.o gindatapage.o \
  	ginbtree.o ginscan.o ginget.o ginvacuum.o ginarrayproc.o \
! 	ginbulk.o ginfast.o
  
  include $(top_srcdir)/src/backend/common.mk
diff -c -r -N src.orig/backend/access/gin/ginbulk.c src/backend/access/gin/ginbulk.c
*** src.orig/backend/access/gin/ginbulk.c	2009-02-04 18:13:28.000000000 +0300
--- src/backend/access/gin/ginbulk.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 197,202 ****
--- 197,204 ----
  	if (nentry <= 0)
  		return;
  
+ 	Assert(ItemPointerIsValid(heapptr) && attnum >= FirstOffsetNumber);
+ 
  	i = nentry - 1;
  	for (; i > 0; i >>= 1)
  		nbit++;
diff -c -r -N src.orig/backend/access/gin/gindatapage.c src/backend/access/gin/gindatapage.c
*** src.orig/backend/access/gin/gindatapage.c	2009-02-04 18:13:28.000000000 +0300
--- src/backend/access/gin/gindatapage.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 43,50 ****
  
  	while (aptr - a < na && bptr - b < nb)
  	{
! 		if (compareItemPointers(aptr, bptr) > 0)
  			*dptr++ = *bptr++;
  		else
  			*dptr++ = *aptr++;
  	}
--- 43,56 ----
  
  	while (aptr - a < na && bptr - b < nb)
  	{
! 		int cmp = compareItemPointers(aptr, bptr); 
! 		if (cmp > 0)
  			*dptr++ = *bptr++;
+ 		else if ( cmp == 0 )
+ 		{
+ 			*dptr++ = *bptr++;
+ 			aptr++;
+ 		}
  		else
  			*dptr++ = *aptr++;
  	}
***************
*** 630,640 ****
  		gdi->stack = ginFindLeafPage(&gdi->btree, gdi->stack);
  
  		if (gdi->btree.findItem(&(gdi->btree), gdi->stack))
! 			elog(ERROR, "item pointer (%u,%d) already exists",
! 			ItemPointerGetBlockNumber(gdi->btree.items + gdi->btree.curitem),
! 				 ItemPointerGetOffsetNumber(gdi->btree.items + gdi->btree.curitem));
! 
! 		ginInsertValue(&(gdi->btree), gdi->stack);
  
  		gdi->stack = NULL;
  	}
--- 636,651 ----
  		gdi->stack = ginFindLeafPage(&gdi->btree, gdi->stack);
  
  		if (gdi->btree.findItem(&(gdi->btree), gdi->stack))
! 		{
! 			/*
! 			 * gdi->btree.items[ gdi->btree.curitem ] already exists in index
! 			 */
! 			gdi->btree.curitem ++;
! 			LockBuffer(gdi->stack->buffer, GIN_UNLOCK);
! 			freeGinBtreeStack(gdi->stack);
! 		}
! 		else
! 			ginInsertValue(&(gdi->btree), gdi->stack);
  
  		gdi->stack = NULL;
  	}
diff -c -r -N src.orig/backend/access/gin/ginfast.c src/backend/access/gin/ginfast.c
*** src.orig/backend/access/gin/ginfast.c	1970-01-01 03:00:00.000000000 +0300
--- src/backend/access/gin/ginfast.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 0 ****
--- 1,811 ----
+ /*-------------------------------------------------------------------------
+  *
+  * ginfast.c
+  *    Fast insert routines for the Postgres inverted index access method.
+  *    Pending entries are stored in linear list of pages and vacuum 
+  *    will transfer them into regular structure.
+  *
+  * Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
+  * Portions Copyright (c) 1994, Regents of the University of California
+  *
+  * IDENTIFICATION
+  *			$PostgreSQL$
+  *
+  *-------------------------------------------------------------------------
+  */
+ 
+ #include "postgres.h"
+ 
+ #include "access/genam.h"
+ #include "access/gin.h"
+ #include "access/tuptoaster.h"
+ #include "catalog/index.h"
+ #include "commands/vacuum.h"
+ #include "miscadmin.h"
+ #include "storage/bufmgr.h"
+ #include "utils/memutils.h"
+ 
+ 
+ static int32
+ writeListPage(Relation index, Buffer buffer, IndexTuple *tuples, int32 ntuples, BlockNumber rightlink)
+ {
+ 	Page			page = BufferGetPage(buffer);
+ 	int 			i, freesize, size=0;
+ 	OffsetNumber	l, off;
+ 
+ 	START_CRIT_SECTION();
+ 
+ 	GinInitBuffer(buffer, GIN_LIST);
+ 
+ 	off = FirstOffsetNumber;
+ 
+ 	for(i=0; i<ntuples; i++)
+ 	{	
+ 		size += IndexTupleSize(tuples[i]);
+ 		l = PageAddItem(page, (Item)tuples[i], IndexTupleSize(tuples[i]), off, false, false);
+ 
+ 		if (l == InvalidOffsetNumber)
+ 			elog(ERROR, "failed to add item to index page in \"%s\"",
+ 					 RelationGetRelationName(index));
+ 
+ 		off++;
+ 	}
+ 
+ 	GinPageGetOpaque(page)->rightlink = rightlink;
+ 	/*
+ 	 * tail page may contain only the whole row(s) or final
+ 	 * part of row placed on previous pages
+ 	 */
+ 	if ( rightlink == InvalidBlockNumber )
+ 	{
+ 		GinPageSetFullRow(page);
+ 		GinPageGetOpaque(page)->maxoff = 1;
+ 	}
+ 	else
+ 	{
+ 		GinPageGetOpaque(page)->maxoff = 0;
+ 	}
+ 
+ 	freesize = PageGetFreeSpace(page);
+ 
+ 	MarkBufferDirty(buffer);
+ 
+ 	if (!index->rd_istemp)
+ 	{
+ 		XLogRecData				rdata[2];
+ 		ginxlogInsertListPage	data;
+ 		XLogRecPtr  			recptr;
+ 		char	   			   *ptr;
+ 
+ 		rdata[0].buffer = buffer;
+ 		rdata[0].buffer_std = true;
+ 		rdata[0].data = (char*)&data;
+ 		rdata[0].len = sizeof(ginxlogInsertListPage);
+ 		rdata[0].next = rdata+1;
+ 
+ 		rdata[1].buffer = InvalidBuffer;
+ 		ptr = rdata[1].data = palloc( size );
+ 		rdata[1].len = size;
+ 		rdata[1].next = NULL;
+ 
+ 		for(i=0; i<ntuples; i++)
+ 		{
+ 			memcpy(ptr, tuples[i], IndexTupleSize(tuples[i]));
+ 			ptr += IndexTupleSize(tuples[i]);
+ 		}
+ 
+ 		data.blkno = BufferGetBlockNumber(buffer);
+ 		data.rightlink = rightlink;
+ 		data.ntuples = ntuples;
+ 
+ 		recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_INSERT_LISTPAGE, rdata);
+ 		PageSetLSN(page, recptr);
+ 		PageSetTLI(page, ThisTimeLineID);
+ 	}
+ 
+ 	UnlockReleaseBuffer(buffer);
+ 
+ 	END_CRIT_SECTION();
+ 
+ 	return freesize;
+ }
+ 
+ static void
+ makeSublist(Relation index, IndexTuple *tuples, int32 ntuples, GinMetaPageData *res)
+ {
+ 	Buffer			curBuffer = InvalidBuffer;
+ 	Buffer			prevBuffer = InvalidBuffer;
+ 	int 			i, size = 0, tupsize;
+ 	int 			startTuple = 0;
+ 
+ 	Assert(ntuples > 0);
+ 
+ 	/*
+ 	 * Split tuples for pages
+ 	 */
+ 	for(i=0;i<ntuples;i++)
+ 	{
+ 		if ( curBuffer == InvalidBuffer )
+ 		{
+ 			curBuffer = GinNewBuffer(index);
+ 
+ 			if ( prevBuffer != InvalidBuffer )
+ 			{
+ 				res->nPendingPages++;
+ 				writeListPage(index, prevBuffer, tuples+startTuple, i-startTuple, BufferGetBlockNumber(curBuffer));
+ 			}
+ 			else
+ 			{
+ 				res->head = BufferGetBlockNumber(curBuffer);
+ 			}
+ 
+ 			prevBuffer = curBuffer;;
+ 			startTuple = i;
+ 			size = 0;
+ 		}
+ 
+ 		tupsize = IndexTupleSize(tuples[i]) + sizeof(ItemIdData);
+ 
+ 		if ( size + tupsize >= GinListPageSize )
+ 		{
+ 			i--;
+ 			curBuffer = InvalidBuffer;
+ 		}
+ 		else
+ 		{
+ 			size += tupsize;
+ 		}
+ 	}
+ 
+ 	/*
+ 	 * Write last page
+ 	 */
+ 	res->tail = BufferGetBlockNumber(curBuffer);
+ 	res->tailFreeSize = writeListPage(index, curBuffer, tuples+startTuple, ntuples-startTuple, InvalidBlockNumber);
+ 	res->nPendingPages++;
+ 	/* that was only one heap tuple */
+ 	res->nPendingHeapTuples = 1; 
+ }
+ 
+ #define GIN_PAGE_FREESIZE \
+ 	( BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(GinPageOpaqueData)) ) 
+ /*
+  * Inserts collected values during normal insertion. Function guarantees
+  * that all values of heap will be stored sequentually with
+  * preserving order
+  */
+ void
+ ginHeapTupleFastInsert(Relation index, GinState *ginstate, GinTupleCollector *collector)
+ {
+ 	Buffer				metabuffer;
+ 	Page				metapage;
+ 	GinMetaPageData    *metadata = NULL;
+ 	XLogRecData			rdata[2];
+ 	Buffer				buffer = InvalidBuffer;
+ 	Page				page = NULL;
+ 	ginxlogUpdateMeta	data;
+ 	bool				separateList = false;
+ 	bool				needCleanup = false;
+ 
+ 	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(ginxlogUpdateMeta);
+ 	rdata[0].next = NULL;
+ 
+ 	metabuffer = ReadBuffer(index, GIN_METAPAGE_BLKNO);
+ 	metapage = BufferGetPage(metabuffer);
+ 
+ 	if ( collector->sumsize + collector->ntuples * sizeof(ItemIdData) > GIN_PAGE_FREESIZE )
+ 	{
+ 		/*
+ 		 * Total size is greater than one page => make sublist
+ 		 */
+ 		separateList = true;
+ 	}
+ 	else
+ 	{
+ 		LockBuffer(metabuffer, GIN_EXCLUSIVE);
+ 		metadata = GinPageGetMeta(metapage);
+ 
+ 		if ( metadata->head == InvalidBlockNumber || 
+ 			collector->sumsize + collector->ntuples * sizeof(ItemIdData) > metadata->tailFreeSize )
+ 		{
+ 			/*
+ 			 * Pending list is empty or total size is greater than freespace 
+ 			 * on tail page => make sublist
+ 			 * We unlock metabuffer to keep high concurrency
+ 			 */
+ 			separateList = true;
+ 			LockBuffer(metabuffer, GIN_UNLOCK);
+ 		}
+ 	}
+ 
+ 	if ( separateList )
+ 	{
+ 		GinMetaPageData		sublist;
+ 
+ 		/*
+ 		 * We should make sublist separately and append it to the tail
+ 		 */
+ 		memset( &sublist, 0, sizeof(GinMetaPageData) );
+ 
+ 		makeSublist(index, collector->tuples, collector->ntuples, &sublist); 
+ 
+ 		/*
+ 		 * metapage was unlocked, see above
+ 		 */
+ 		LockBuffer(metabuffer, GIN_EXCLUSIVE);
+ 		metadata = GinPageGetMeta(metapage);
+ 
+ 		if ( metadata->head == InvalidBlockNumber )
+ 		{
+ 			/*
+ 			 * Sublist becomes main list
+ 			 */
+ 			START_CRIT_SECTION();
+ 			memcpy(metadata, &sublist, sizeof(GinMetaPageData) );
+ 			memcpy(&data.metadata, &sublist, sizeof(GinMetaPageData) );
+ 		}
+ 		else
+ 		{
+ 			/*
+ 			 * merge lists
+ 			 */
+ 
+ 			data.prevTail = metadata->tail;
+ 			buffer = ReadBuffer(index, metadata->tail);
+ 			LockBuffer(buffer, GIN_EXCLUSIVE);
+ 			page = BufferGetPage(buffer);
+ 			Assert(GinPageGetOpaque(page)->rightlink == InvalidBlockNumber);
+ 
+ 			START_CRIT_SECTION();
+ 
+ 			GinPageGetOpaque(page)->rightlink = sublist.head;
+ 			metadata->tail = sublist.tail;
+ 			metadata->tailFreeSize = sublist.tailFreeSize;
+ 
+ 			metadata->nPendingPages += sublist.nPendingPages;
+ 			metadata->nPendingHeapTuples += sublist.nPendingHeapTuples;
+ 
+ 			memcpy(&data.metadata, metadata, sizeof(GinMetaPageData) );
+ 			data.newRightlink = sublist.head;
+ 
+ 			MarkBufferDirty(buffer);
+ 		}
+ 	}
+ 	else
+ 	{
+ 		/*
+ 		 * Insert into tail page, metapage is already locked
+ 		 */
+ 
+ 		OffsetNumber	l, off;
+ 		int				i, tupsize;
+ 		char			*ptr;
+ 
+ 		buffer = ReadBuffer(index, metadata->tail);
+ 		LockBuffer(buffer, GIN_EXCLUSIVE);
+ 		page = BufferGetPage(buffer);
+ 		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
+ 		 */
+ 		Assert( GinPageGetOpaque(page)->maxoff <= metadata->nPendingHeapTuples );
+ 		GinPageGetOpaque(page)->maxoff++;
+ 		metadata->nPendingHeapTuples++;
+ 
+ 		for(i=0; i<collector->ntuples; i++)
+ 		{
+ 			tupsize = IndexTupleSize(collector->tuples[i]);
+ 			l = PageAddItem(page, (Item)collector->tuples[i], tupsize, off, false, false);
+ 
+ 			if (l == InvalidOffsetNumber)
+ 				elog(ERROR, "failed to add item to index page in \"%s\"",
+ 						 RelationGetRelationName(index));
+ 
+ 			memcpy(ptr, collector->tuples[i], tupsize);
+ 			ptr+=tupsize;
+ 
+ 			off++;
+ 		}
+ 
+ 		metadata->tailFreeSize -= collector->sumsize + collector->ntuples * sizeof(ItemIdData);
+ 		memcpy(&data.metadata, metadata, sizeof(GinMetaPageData) );
+ 		MarkBufferDirty(buffer);
+ 	}
+ 
+ 	/*
+ 	 *  Make real write 
+ 	 */
+ 
+ 	MarkBufferDirty(metabuffer);
+ 	if ( !index->rd_istemp )
+ 	{
+ 		XLogRecPtr  recptr;
+ 
+ 		recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_UPDATE_META_PAGE, rdata);
+ 		PageSetLSN(metapage, recptr);
+ 		PageSetTLI(metapage, ThisTimeLineID);
+ 
+ 		if ( buffer != InvalidBuffer )
+ 		{
+ 			PageSetLSN(page, recptr);
+ 			PageSetTLI(page, ThisTimeLineID);
+ 		}
+ 	}
+ 
+ 	if (buffer != InvalidBuffer)
+ 		UnlockReleaseBuffer(buffer);
+ 
+ 	/*
+ 	 * Force pending list cleanup when it becomes too long
+ 	 * to keep non-lossy tidbitmap during pending list scan.
+ 	 * Also, ginInsertCleanup could take significant amount of
+ 	 * time, so we prefer call it when it can do all things by 
+ 	 * single collection cycle. In non-vacuum mode, it shouldn't
+ 	 * require maintenance_work_mem, so let we fire a trigger
+ 	 * while pending list is small enough to fit into work_mem
+ 	 *
+ 	 * ginInsertCleanup() should not be called inside CRIT_SECTION.
+ 	 */
+ 
+ 	if ( metadata->nPendingPages * GIN_PAGE_FREESIZE > work_mem * 1024L ||
+ 		 metadata->nPendingHeapTuples >= tbm_max_non_lossy( work_mem * 1024L ) )
+ 	{
+ 		needCleanup = true;
+ 	}
+ 
+ 	UnlockReleaseBuffer(metabuffer);
+ 
+ 	END_CRIT_SECTION();
+ 
+ 	if ( needCleanup )
+ 	{
+ 		ginInsertCleanup(index, ginstate, NULL);
+ 	}
+ }
+ 
+ /*
+  * Collect values from one tuples to be indexed. All values for
+  * one tuples should be written at once - to guarantee consistent state
+  */
+ uint32
+ ginHeapTupleFastCollect(Relation index, GinState *ginstate, GinTupleCollector *collector,
+ 		OffsetNumber attnum, Datum value, ItemPointer item)
+ {
+ 	Datum	   *entries;
+ 	int32		i,
+ 				nentries;
+ 
+ 	entries = extractEntriesSU(ginstate, attnum, value, &nentries);
+ 
+ 	if (nentries == 0)
+ 		/* nothing to insert */
+ 		return 0;
+ 
+ 	/*
+ 	 * Allocate/reallocate memory for storing collected tuples
+ 	 */
+ 	if ( collector->tuples == NULL )
+ 	{
+ 		collector->lentuples = nentries * index->rd_att->natts;
+ 		collector->tuples = (IndexTuple*)palloc(sizeof(IndexTuple) * collector->lentuples);
+ 	}
+ 	
+ 	while ( collector->ntuples + nentries > collector->lentuples )
+ 	{
+ 		collector->lentuples *= 2;
+ 		collector->tuples = (IndexTuple*)repalloc( collector->tuples,
+ 													sizeof(IndexTuple) * collector->lentuples);
+ 	}
+ 
+ 	/*
+ 	 * Creates tuple's array
+ 	 */
+ 	for (i = 0; i < nentries; i++)
+ 	{
+ 		int32 tupsize;
+ 
+ 		collector->tuples[collector->ntuples + i] = GinFormTuple(ginstate, attnum, entries[i], NULL, 0);
+ 		collector->tuples[collector->ntuples + i]->t_tid = *item;
+ 		tupsize = IndexTupleSize(collector->tuples[collector->ntuples + i]);
+ 
+ 		if ( tupsize > TOAST_INDEX_TARGET || tupsize >= GinMaxItemSize)
+ 			elog(ERROR, "huge tuple");
+ 
+ 		collector->sumsize += tupsize;
+ 	}
+ 
+ 	collector->ntuples += nentries;
+ 
+ 	return nentries;
+ }
+ 
+ /*
+  * Deletes first pages in list before newHead page.
+  * If newHead == InvalidBlockNumber then function drops the whole list.
+  * returns true if concurrent completion process is running
+  */
+ static bool
+ shiftList(Relation index, Buffer metabuffer, BlockNumber newHead,
+ 		  IndexBulkDeleteResult *stats)
+ {
+ #define	NDELETE_AT_ONCE	(16)
+ 	Buffer					buffers[NDELETE_AT_ONCE];
+ 	ginxlogDeleteListPages	data;
+ 	XLogRecData				rdata[1];
+ 	Page					metapage;
+ 	GinMetaPageData 	   *metadata;
+ 	BlockNumber				blknoToDelete;
+ 
+ 	metapage = BufferGetPage(metabuffer);
+ 	metadata = GinPageGetMeta(metapage);
+ 	blknoToDelete = metadata->head;
+ 
+ 	data.node = index->rd_node;
+ 
+ 	rdata[0].buffer = InvalidBuffer;
+ 	rdata[0].data = (char *) &data;
+ 	rdata[0].len = sizeof(ginxlogDeleteListPages);
+ 	rdata[0].next = NULL;
+ 
+ 	do
+ 	{
+ 		Page		page;
+ 		int			i;
+ 		int64		nDeletedHeapTuples = 0;
+ 
+ 		data.ndeleted = 0;
+ 		while( data.ndeleted < NDELETE_AT_ONCE && blknoToDelete != newHead )
+ 		{
+ 			data.toDelete[ data.ndeleted ] = blknoToDelete; 
+ 			buffers[ data.ndeleted ] = ReadBuffer(index, blknoToDelete);
+ 			LockBufferForCleanup( buffers[ data.ndeleted ] );
+ 			page = BufferGetPage( buffers[ data.ndeleted ] );
+ 
+ 			data.ndeleted++;
+ 			if (stats)
+ 				stats->pages_deleted++;
+ 
+ 			if ( GinPageIsDeleted(page) )
+ 			{
+ 				/* concurrent deletion process is detected */
+ 				for(i=0;i<data.ndeleted;i++)
+ 					UnlockReleaseBuffer( buffers[i] );
+ 
+ 				return true;
+ 			}
+ 
+ 			nDeletedHeapTuples += GinPageGetOpaque(page)->maxoff;
+ 			blknoToDelete = GinPageGetOpaque( page )->rightlink;
+ 		}
+ 
+ 		START_CRIT_SECTION();
+ 		
+ 		metadata->head = blknoToDelete;
+ 
+ 		Assert( metadata->nPendingPages >= data.ndeleted );
+ 		metadata->nPendingPages -= data.ndeleted;
+ 		Assert( metadata->nPendingHeapTuples >= nDeletedHeapTuples );
+ 		metadata->nPendingHeapTuples -= nDeletedHeapTuples;
+ 
+ 		if ( blknoToDelete == InvalidBlockNumber )
+ 		{
+ 			metadata->tail = InvalidBlockNumber;
+ 			metadata->tailFreeSize = 0;
+ 			metadata->nPendingPages = 0;
+ 			metadata->nPendingHeapTuples = 0;
+ 		}
+ 		memcpy( &data.metadata, metadata, sizeof(GinMetaPageData));
+ 		MarkBufferDirty( metabuffer );
+ 
+ 		for(i=0; i<data.ndeleted; i++)
+ 		{
+ 			page = BufferGetPage( buffers[ i ] );
+ 			GinPageGetOpaque( page )->flags = GIN_DELETED;
+ 			MarkBufferDirty( buffers[ i ] );
+ 		}
+ 
+ 		if ( !index->rd_istemp )
+ 		{
+ 			XLogRecPtr  recptr;
+ 
+ 			recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_DELETE_LISTPAGE, rdata);
+ 			PageSetLSN(metapage, recptr);
+ 			PageSetTLI(metapage, ThisTimeLineID);
+ 
+ 			for(i=0; i<data.ndeleted; i++)
+ 			{
+ 				page = BufferGetPage( buffers[ i ] );
+ 				PageSetLSN(page, recptr);
+ 				PageSetTLI(page, ThisTimeLineID);
+ 			}
+ 		}
+ 
+ 		for(i=0; i<data.ndeleted; i++)
+ 			UnlockReleaseBuffer( buffers[ i ] );
+ 
+ 		END_CRIT_SECTION();
+ 	} while( blknoToDelete != newHead );
+ 
+ 	return false;
+ }
+ 
+ typedef struct DatumArray
+ {
+ 	Datum	*values;
+ 	int32	 nvalues;
+ 	int32	 maxvalues;
+ } DatumArray;
+ 
+ static void 
+ addDatum(DatumArray *datums, Datum datum)
+ {
+ 	if ( datums->nvalues >= datums->maxvalues)
+ 	{
+ 		datums->maxvalues *= 2;
+ 		datums->values = (Datum*)repalloc( datums->values, sizeof(Datum)*datums->maxvalues);
+ 	}
+ 
+ 	datums->values[ datums->nvalues++ ] = datum; 
+ }
+ 
+ /*
+  * Go through all tuples on page and collect values in memory
+  */
+ 
+ static void 
+ processPendingPage(BuildAccumulator *accum, DatumArray *da, Page page, OffsetNumber startoff)
+ {
+ 	ItemPointerData	heapptr;
+ 	OffsetNumber 	i,maxoff;
+ 	OffsetNumber	attrnum, curattnum;
+ 
+ 	maxoff = PageGetMaxOffsetNumber(page);
+ 	Assert( maxoff >= FirstOffsetNumber );
+ 	ItemPointerSetInvalid(&heapptr);
+ 	attrnum = 0;
+ 
+ 	for (i = startoff; i <= maxoff; i = OffsetNumberNext(i))
+ 	{
+ 		IndexTuple  itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, i));
+ 
+ 		curattnum = gintuple_get_attrnum(accum->ginstate, itup);
+ 
+ 		if ( !ItemPointerIsValid(&heapptr) )
+ 		{
+ 			heapptr = itup->t_tid;
+ 			attrnum = curattnum;
+ 		}
+ 		else if ( !(ItemPointerEquals(&heapptr, &itup->t_tid) && curattnum == attrnum) )
+ 		{
+ 			/*
+ 			 * We can insert several datums per call, but only for one heap tuple
+ 			 * and one column.
+ 			 */
+ 			ginInsertRecordBA(accum, &heapptr, attrnum, da->values, da->nvalues);
+ 			da->nvalues = 0;
+ 			heapptr = itup->t_tid;
+ 			attrnum = curattnum;
+ 		}
+ 		addDatum(da, gin_index_getattr(accum->ginstate, itup));
+ 	}
+ 
+ 	ginInsertRecordBA(accum, &heapptr, attrnum, da->values, da->nvalues);
+ }
+ 
+ /*
+  * Moves tuples from pending pages into regular GIN structure.
+  * Function doesn't require special locking and could be called
+  * in any time but only one at the same time.
+  *
+  * Non-NULL stats indicates that ginInsertCleanup is called
+  * from vacuum process, so call vacuum_delay_point() periodically.
+  */ 
+ 
+ void 
+ ginInsertCleanup(Relation index, GinState *ginstate, IndexBulkDeleteResult *stats)
+ {
+ 	Buffer				metabuffer, buffer;
+ 	Page				metapage, page;
+ 	GinMetaPageData    *metadata;
+ 	MemoryContext		opCtx, oldCtx;
+ 	BuildAccumulator	accum;
+ 	DatumArray			datums;
+ 	BlockNumber			blkno;
+ 
+ 	metabuffer = ReadBuffer(index, GIN_METAPAGE_BLKNO);
+ 	LockBuffer(metabuffer, GIN_SHARE);
+ 	metapage = BufferGetPage(metabuffer);
+ 	metadata = GinPageGetMeta(metapage);
+ 
+ 	if ( metadata->head == InvalidBlockNumber )
+ 	{
+ 		UnlockReleaseBuffer(metabuffer);
+ 		return;
+ 	}
+ 
+ 	/*
+ 	 * Init
+ 	 */
+ 	datums.maxvalues=128;
+ 	datums.nvalues = 0;
+ 	datums.values = (Datum*)palloc(sizeof(Datum)*datums.maxvalues);
+ 
+ 	ginInitBA(&accum);
+ 	accum.ginstate = ginstate;
+ 
+ 	opCtx = AllocSetContextCreate(CurrentMemoryContext,
+ 									"Gin refresh temporary context",
+ 									ALLOCSET_DEFAULT_MINSIZE,
+ 									ALLOCSET_DEFAULT_INITSIZE,
+ 									ALLOCSET_DEFAULT_MAXSIZE);
+ 
+ 	oldCtx = MemoryContextSwitchTo(opCtx);
+ 
+ 	/*
+ 	 * Read and lock head
+ 	 */
+ 	blkno = metadata->head;
+ 	buffer = ReadBuffer(index, blkno);
+ 	LockBuffer(buffer, GIN_SHARE);
+ 	page = BufferGetPage(buffer);
+ 
+ 	LockBuffer(metabuffer, GIN_UNLOCK);
+ 
+ 	for(;;)
+ 	{
+ 		/*
+ 		 * reset datum's collector and read page's datums into memory
+ 		 */
+ 		datums.nvalues = 0;
+ 
+ 		if ( GinPageIsDeleted(page) )
+ 		{
+ 			/* concurrent completion is running */
+ 			UnlockReleaseBuffer( buffer );
+ 			break;
+ 		}
+ 
+ 		processPendingPage(&accum, &datums, page, FirstOffsetNumber);
+ 
+ 		if (stats)
+ 			vacuum_delay_point();
+ 
+ 		/*
+ 		 * Is it time to flush memory to disk?
+ 		 */
+ 		if ( GinPageGetOpaque(page)->rightlink == InvalidBlockNumber || 
+ 				( GinPageHasFullRow(page) && accum.allocatedMemory > maintenance_work_mem * 1024L ) )
+ 		{
+ 			ItemPointerData    *list;
+ 			uint32      		nlist;
+ 			Datum       		entry;
+ 			OffsetNumber  		maxoff, attnum;
+ 
+ 			/*
+ 			 * Unlock current page to increase performance.
+ 			 * Changes of page will be checked later by comparing 
+ 			 * maxoff after completion of memory flush. 
+ 			 */
+ 			maxoff = PageGetMaxOffsetNumber(page);
+ 			LockBuffer(buffer, GIN_UNLOCK);
+ 
+ 			/*
+ 			 * Moving collected data into regular structure can take
+ 			 * significant amount of time - so, run it without locking pending
+ 			 * list.
+ 			 */
+ 			while ((list = ginGetEntry(&accum, &attnum, &entry, &nlist)) != NULL)
+ 			{
+ 				if (stats)
+ 					vacuum_delay_point();
+ 				ginEntryInsert(index, ginstate, attnum, entry, list, nlist, FALSE);
+ 			}
+ 
+ 			/*
+ 			 * Lock the whole list to remove pages 
+ 			 */
+ 			LockBuffer(metabuffer, GIN_EXCLUSIVE);
+ 			LockBuffer(buffer, GIN_SHARE);
+ 
+ 			if ( GinPageIsDeleted(page) )
+ 			{
+ 				/* concurrent completion is running */
+ 				UnlockReleaseBuffer(buffer);
+ 				LockBuffer(metabuffer, GIN_UNLOCK);
+ 				break;
+ 			}
+ 
+ 			/*
+ 			 * While we keeped page unlocked it might be changed - 
+ 			 * add read the changes separately. On one page is rather
+ 			 * small - so, overused memory isn't very big, although 
+ 			 * we should reinit accumulator. We need to make a
+ 			 * check only once because now both page and metapage are
+ 			 * locked. Inserion algorithm gurantees that inserted row(s)
+ 			 * will not continue on next page.
+ 			 */
+ 			if ( PageGetMaxOffsetNumber(page) != maxoff )
+ 			{
+ 				ginInitBA(&accum);
+ 				datums.nvalues = 0;
+ 				processPendingPage(&accum, &datums, page, maxoff+1);
+ 
+ 				while ((list = ginGetEntry(&accum, &attnum, &entry, &nlist)) != NULL)
+ 					ginEntryInsert(index, ginstate, attnum, entry, list, nlist, FALSE);
+ 			}
+ 
+ 			/*
+ 			 * Remember next page - it will become a new head
+ 			 */
+ 			blkno = GinPageGetOpaque(page)->rightlink;
+ 			UnlockReleaseBuffer(buffer); /* shiftList will do exclusive locking */
+ 
+ 			/*
+ 			 * remove readed pages from pending list, at this point all 
+ 			 * content of readed pages is in regular structure
+ 			 */
+ 			if ( shiftList(index, metabuffer, blkno, stats) )
+ 			{
+ 				/* concurrent completion is running */
+ 				LockBuffer(metabuffer, GIN_UNLOCK);
+ 				break;
+ 			}
+ 
+ 			Assert( blkno == metadata->head );
+ 			LockBuffer(metabuffer, GIN_UNLOCK);
+ 
+ 			/*
+ 			 * if we remove the whole list just exit 
+ 			 */
+ 			if ( blkno == InvalidBlockNumber )
+ 				break;
+ 
+ 			/*
+ 			 * reinit state
+ 			 */
+ 			MemoryContextReset(opCtx);
+ 			ginInitBA(&accum);
+ 		}
+ 		else
+ 		{
+ 			blkno = GinPageGetOpaque(page)->rightlink;
+ 			UnlockReleaseBuffer(buffer);
+ 		}
+ 
+ 		/*
+ 		 * Read next page in pending list
+ 		 */
+ 		CHECK_FOR_INTERRUPTS();
+ 		buffer = ReadBuffer(index, blkno);
+ 		LockBuffer(buffer, GIN_SHARE);
+ 		page = BufferGetPage(buffer);
+ 	}
+ 
+ 	ReleaseBuffer(metabuffer);
+ 	MemoryContextSwitchTo(oldCtx);
+ 	MemoryContextDelete(opCtx);
+ }
+ 
diff -c -r -N src.orig/backend/access/gin/ginget.c src/backend/access/gin/ginget.c
*** src.orig/backend/access/gin/ginget.c	2009-02-04 18:13:28.000000000 +0300
--- src/backend/access/gin/ginget.c	2009-02-04 18:33:01.000000000 +0300
***************
*** 268,273 ****
--- 268,282 ----
  	Page			page;
  	bool			needUnlock = TRUE;
  
+ 	entry->buffer = InvalidBuffer;
+ 	entry->offset = InvalidOffsetNumber;
+ 	entry->list = NULL;
+ 	entry->nlist = 0;
+ 	entry->partialMatch = NULL;
+ 	entry->partialMatchResult = NULL;
+ 	entry->reduceResult = FALSE;
+ 	entry->predictNumberResult = 0;
+ 
  	if (entry->master != NULL)
  	{
  		entry->isFinished = entry->master->isFinished;
***************
*** 285,299 ****
  	page = BufferGetPage(stackEntry->buffer);
  
  	entry->isFinished = TRUE;
- 	entry->buffer = InvalidBuffer;
- 	entry->offset = InvalidOffsetNumber;
- 	entry->list = NULL;
- 	entry->nlist = 0;
- 	entry->partialMatch = NULL;
- 	entry->partialMatchIterator = NULL;
- 	entry->partialMatchResult = NULL;
- 	entry->reduceResult = FALSE;
- 	entry->predictNumberResult = 0;
  
  	if ( entry->isPartialMatch )
  	{
--- 294,299 ----
***************
*** 354,362 ****
  
  			entry->buffer = scanBeginPostingTree(gdi);
  			/*
! 			 * We keep buffer pinned because we need to prevent deletition
  			 * page during scan. See GIN's vacuum implementation. RefCount
! 			 * is increased to keep buffer pinned after freeGinBtreeStack() call.
  			 */
  			IncrBufferRefCount(entry->buffer);
  
--- 354,363 ----
  
  			entry->buffer = scanBeginPostingTree(gdi);
  			/*
! 			 * We keep buffer pinned because we need to prevent deletion of
  			 * page during scan. See GIN's vacuum implementation. RefCount
! 			 * is increased to keep buffer pinned after freeGinBtreeStack()
! 			 * call.
  			 */
  			IncrBufferRefCount(entry->buffer);
  
***************
*** 433,438 ****
--- 434,448 ----
  	uint32		i;
  	GinScanOpaque so = (GinScanOpaque) scan->opaque;
  
+ 	/*
+ 	 * If isScanFastInsert is still true, set up to scan the pending-insert
+ 	 * list rather than the main index.
+ 	 */
+ 	if (so->isScanFastInsert)
+ 	{
+ 		return;
+ 	}
+ 
  	for (i = 0; i < so->nkeys; i++)
  		startScanKey(scan->indexRelation, &so->ginstate, so->keys + i);
  }
***************
*** 677,682 ****
--- 687,1032 ----
  	return FALSE;
  }
  
+ typedef struct fastPosition {
+ 	Buffer			fastBuffer;
+ 	OffsetNumber	firstOffset;
+ 	OffsetNumber	lastOffset;
+ 	ItemPointerData	item;
+ } fastPosition;
+ 
+ 
+ /*
+  * Get ItemPointer of next heap row to be checked from fast insert storage.
+  * Returns false if there are no more.
+  *
+  * The fastBuffer is presumed pinned and share-locked on entry, and is 
+  * pinned and share-locked on success exit.  On failure exit it's released.
+  */
+ static bool
+ scanGetCandidate(IndexScanDesc scan, fastPosition *pos)
+ {
+ 	OffsetNumber		maxoff;
+ 	Page				page;
+ 	IndexTuple  		itup;
+ 
+ 	ItemPointerSetInvalid( &pos->item );
+ 	for(;;)
+ 	{
+ 		page = BufferGetPage(pos->fastBuffer);
+ 
+ 		maxoff = PageGetMaxOffsetNumber(page);
+ 		if ( pos->firstOffset > maxoff )
+ 		{
+ 			BlockNumber blkno = GinPageGetOpaque(page)->rightlink;
+ 			if ( blkno == InvalidBlockNumber )
+ 			{
+ 				UnlockReleaseBuffer(pos->fastBuffer);
+ 				pos->fastBuffer=InvalidBuffer;
+ 
+ 				return false;
+ 			}
+ 			else
+ 			{
+ 				/*
+ 				 * Here we should prevent deletion of next page by 
+ 				 * insertcleanup process, which uses LockBufferForCleanup.
+ 				 * So, we pin next page before unpin current one
+ 				 */
+ 				Buffer	tmpbuf = ReadBuffer(scan->indexRelation, blkno);
+ 
+ 				UnlockReleaseBuffer( pos->fastBuffer);
+ 				pos->fastBuffer=tmpbuf;
+ 				LockBuffer( pos->fastBuffer, GIN_SHARE );
+ 
+ 				pos->firstOffset = FirstOffsetNumber;
+ 			}
+ 		}
+ 		else
+ 		{
+ 			itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, pos->firstOffset));
+ 			pos->item = itup->t_tid;
+ 			if ( GinPageHasFullRow(page) )
+ 			{
+ 				/*
+ 				 * find itempointer to the next row
+ 				 */
+ 				for(pos->lastOffset = pos->firstOffset+1; pos->lastOffset<=maxoff; pos->lastOffset++)
+ 				{
+ 					itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, pos->lastOffset));
+ 					if (!ItemPointerEquals(&pos->item, &itup->t_tid))
+ 						break;
+ 				}
+ 			}
+ 			else
+ 			{
+ 				/*
+ 				 * All itempointers are the same on this page
+ 				 */
+ 				pos->lastOffset = maxoff + 1;
+ 			}
+ 			break;
+ 		}
+ 	}
+ 
+ 	return true;
+ }
+ 
+ static bool
+ matchPartialInPendingList(GinState *ginstate, Page page, OffsetNumber off, 
+ 	OffsetNumber maxoff, Datum value, OffsetNumber attrnum,
+ 	Datum	*datum, bool *datumExtracted, StrategyNumber strategy)
+ {
+ 	IndexTuple  		itup;
+ 	int 				res;
+ 
+ 	while( off < maxoff )
+ 	{
+ 		itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, off));
+ 		if ( attrnum != gintuple_get_attrnum(ginstate, itup) )
+ 			return false;
+ 		
+ 		if (datumExtracted[ off-1 ] == false)
+ 		{
+ 			datum[ off-1 ] = gin_index_getattr(ginstate, itup);
+ 			datumExtracted[  off-1 ] = true;
+ 		}
+ 
+ 		res = DatumGetInt32(FunctionCall3(&ginstate->comparePartialFn[attrnum],
+ 						  value,
+ 						  datum[ off-1 ],
+ 						  UInt16GetDatum(strategy)));
+ 		if ( res == 0 )
+ 			return true;
+ 		else if (res>0)
+ 			return false;
+ 	}
+ 
+ 	return false;
+ }
+ /*
+  * Sets entryRes array for each key by looking on
+  * every entry per indexed value (row) in fast insert storage.
+  * returns true if at least one of datum was matched by key's entry
+  *
+  * The fastBuffer is presumed pinned and share-locked on entry.
+  */
+ static bool
+ collectDatumForItem(IndexScanDesc scan, fastPosition *pos)
+ {
+ 	GinScanOpaque 		so = (GinScanOpaque) scan->opaque;
+ 	OffsetNumber		attrnum;
+ 	Page				page;
+ 	IndexTuple  		itup;
+ 	int					i, j;
+ 	bool				hasMatch = false;
+ 
+ 	/*
+ 	 * Resets entryRes
+ 	 */
+ 	for (i = 0; i < so->nkeys; i++)
+ 	{
+ 		GinScanKey	key = so->keys + i;
+ 		memset( key->entryRes, FALSE, key->nentries );
+ 	}
+ 
+ 	for(;;)
+ 	{
+ 		Datum				datum[ BLCKSZ/sizeof(IndexTupleData) ];
+ 		bool				datumExtracted[ BLCKSZ/sizeof(IndexTupleData) ];
+ 
+ 		Assert( pos->lastOffset > pos->firstOffset );
+ 		memset(datumExtracted + pos->firstOffset - 1, 0, sizeof(bool) * (pos->lastOffset - pos->firstOffset ));
+ 
+ 		page = BufferGetPage(pos->fastBuffer);
+ 
+ 		for(i = 0; i < so->nkeys; i++)
+ 		{
+ 			GinScanKey  key = so->keys + i;
+ 
+ 			for(j=0; j<key->nentries; j++)
+ 			{
+ 				OffsetNumber		StopLow = pos->firstOffset,
+ 									StopHigh = pos->lastOffset,
+ 									StopMiddle;
+ 				GinScanEntry		entry = key->scanEntry + j;
+ 
+ 				if ( key->entryRes[j] )
+ 					continue;
+ 
+ 				while (StopLow < StopHigh)
+ 				{
+ 					StopMiddle = StopLow + ((StopHigh - StopLow) >> 1);
+ 
+ 					itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, StopMiddle));
+ 					attrnum = gintuple_get_attrnum(&so->ginstate, itup);
+ 
+ 					if (key->attnum < attrnum)
+ 						StopHigh = StopMiddle;
+ 					else if (key->attnum > attrnum)
+ 						StopLow = StopMiddle + 1;
+ 					else
+ 					{
+ 						int res;
+ 
+ 						if (datumExtracted[ StopMiddle-1 ] == false)
+ 						{
+ 							datum[ StopMiddle-1 ] = gin_index_getattr(&so->ginstate, itup);
+ 							datumExtracted[  StopMiddle-1 ] = true;
+ 						}
+ 						res =  compareEntries(&so->ginstate, 
+ 									entry->attnum, 
+ 									entry->entry, 
+ 									datum[ StopMiddle-1 ]);
+ 
+ 						if ( res == 0 )
+ 						{
+ 							if ( entry->isPartialMatch )
+ 								key->entryRes[j] = matchPartialInPendingList(&so->ginstate, page, StopMiddle, 
+ 																	pos->lastOffset, entry->entry, entry->attnum,
+ 																	datum, datumExtracted, entry->strategy);
+ 							else
+ 								key->entryRes[j] = true;
+ 							break;
+ 						}
+ 						else if ( res < 0  )
+ 							StopHigh = StopMiddle;
+ 						else
+ 							StopLow = StopMiddle + 1;
+ 					}
+ 				}
+ 
+ 				if ( StopLow>=StopHigh && entry->isPartialMatch )
+ 					key->entryRes[j] = matchPartialInPendingList(&so->ginstate, page, StopHigh, 
+ 																	pos->lastOffset, entry->entry, entry->attnum,
+ 																	datum, datumExtracted, entry->strategy);
+ 
+ 				hasMatch |= key->entryRes[j];
+ 			}
+ 		}
+ 	
+ 		pos->firstOffset = pos->lastOffset;
+ 
+ 		if ( GinPageHasFullRow(page) )
+ 		{
+ 			/*
+ 			 * We scan all values from one tuple, go to next one
+ 			 */
+ 
+ 			return hasMatch;
+ 		}
+ 		else
+ 		{
+ 			ItemPointerData item = pos->item;
+ 
+ 			if ( scanGetCandidate(scan, pos) == false || !ItemPointerEquals(&pos->item, &item) )
+ 				elog(ERROR,"Could not process tuple");  /* XXX should not be here ! */
+ 		}
+ 	}
+ 
+ 	return hasMatch;
+ }
+ 
+ /*
+  * Collect all matched rows from pending list in bitmap
+  */
+ static TIDBitmap*
+ scanFastInsert(IndexScanDesc scan, TIDBitmap *tbm, int64 *ntids)
+ {
+ 	GinScanOpaque so = (GinScanOpaque) scan->opaque;
+ 	MemoryContext	oldCtx;
+ 	bool			recheck, keyrecheck, match;
+ 	int				i;
+ 	fastPosition	pos;
+ 	Buffer      	metabuffer = ReadBuffer(scan->indexRelation, GIN_METAPAGE_BLKNO);
+ 	BlockNumber 	blkno;
+ 
+ 	if ( ntids )
+ 		*ntids = 0;
+ 
+ 	LockBuffer(metabuffer, GIN_SHARE);
+ 	blkno = GinPageGetMeta(BufferGetPage(metabuffer))->head;
+ 
+ 	/*
+ 	 * fetch head of list before unlocking metapage.
+ 	 * head page must be pinned to prevent deletion by vacuum process
+ 	 */
+ 	if ( blkno == InvalidBlockNumber )
+ 	{
+ 		/* No pending list, so proceed with normal scan */
+ 		UnlockReleaseBuffer( metabuffer );
+ 		return NULL;
+ 	}
+ 
+ 	pos.fastBuffer = ReadBuffer(scan->indexRelation, blkno);
+ 	LockBuffer(pos.fastBuffer, GIN_SHARE);
+ 	pos.firstOffset = FirstOffsetNumber;
+ 	UnlockReleaseBuffer( metabuffer );
+ 
+ 	/*
+ 	 * loop for each heap row
+ 	 */
+ 	while( scanGetCandidate(scan, &pos) )
+ 	{
+ 
+ 		/*
+ 		 * Check entries in rows and setup entryRes array
+ 		 */
+ 		if (!collectDatumForItem(scan, &pos))
+ 			continue;
+ 
+ 		/*
+ 		 * check for consistent
+ 		 */
+ 		oldCtx = MemoryContextSwitchTo(so->tempCtx);
+ 		recheck = false;
+ 		match = true; 
+ 
+ 		for (i = 0; match && i < so->nkeys; i++)
+ 		{
+ 			GinScanKey  	key = so->keys + i;
+ 
+ 			keyrecheck = true;
+ 
+ 			if ( DatumGetBool(FunctionCall4(&so->ginstate.consistentFn[ key->attnum-1 ],
+ 									 PointerGetDatum(key->entryRes),
+ 									 UInt16GetDatum(key->strategy),
+ 									 key->query,
+ 									 PointerGetDatum(&keyrecheck))) == false )
+ 			{
+ 				match = false;
+ 			}
+ 
+ 			recheck |= keyrecheck;
+ 		}
+ 
+ 		MemoryContextSwitchTo(oldCtx);
+ 		MemoryContextReset(so->tempCtx);
+ 
+ 		if ( match )
+ 		{
+ 			if ( tbm == NULL )
+ 				tbm = tbm_create( work_mem * 1024L );
+ 			tbm_add_tuples(tbm, &pos.item, 1, recheck);
+ 			if ( ntids )
+ 				(*ntids)++;
+ 		}
+ 	}
+ 
+ 	/*
+ 	 * ntids != NULL means gingetbitmap, so we shouldn't
+ 	 * take care of tbm's losiness because bitmap scan is protected
+ 	 * from duplication values in results.
+ 	 */
+ 
+ 	if ( ntids == NULL && tbm && tbm_has_lossy(tbm) )
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_OUT_OF_MEMORY),
+ 					errmsg("not enough memory to store result of pending list or VACUUM table" ),
+ 					errhint("Increase the \"work_mem\" parameter.")));
+ 
+ 	return tbm;
+ }
+ 
  /*
   * Get heap item pointer from scan
   * returns true if found
***************
*** 699,742 ****
  	 */
  	*recheck = false;
  
! 	ItemPointerSetMin(item);
! 	for (i = 0; i < so->nkeys; i++)
  	{
! 		GinScanKey	key = so->keys + i;
  
! 		if (keyGetItem(scan->indexRelation, &so->ginstate, so->tempCtx,
! 					   key, &keyrecheck))
! 			return FALSE;		/* finished one of keys */
! 		if (compareItemPointers(item, &key->curItem) < 0)
! 			*item = key->curItem;
! 		*recheck |= keyrecheck;
! 	}
  
! 	for (i = 1; i <= so->nkeys; i++)
! 	{
! 		GinScanKey	key = so->keys + i - 1;
  
! 		for (;;)
  		{
! 			int			cmp = compareItemPointers(item, &key->curItem);
  
! 			if (cmp == 0)
! 				break;
! 			else if (cmp > 0)
  			{
! 				if (keyGetItem(scan->indexRelation, &so->ginstate, so->tempCtx,
! 							   key, &keyrecheck))
! 					return FALSE;		/* finished one of keys */
! 				*recheck |= keyrecheck;
  			}
  			else
! 			{					/* returns to begin */
  				*item = key->curItem;
! 				i = 0;
  				break;
  			}
  		}
! 	}
  
  	return TRUE;
  }
--- 1049,1162 ----
  	 */
  	*recheck = false;
  
! 	/*
! 	 * First of all we should check fast insert list of pages
! 	 */
! 	if ( so->isScanFastInsert )
  	{
! 		if ( so->scanFastTuples )
! 		{
! 			/*
! 			 * Items from pending list is already collected in memory
! 			 */
  
! 			if ( so->scanFastResult == NULL || so->scanFastOffset >= so->scanFastResult->ntuples )
! 			{
! 				so->scanFastResult = tbm_iterate( so->scanFastIterator );
  
! 				if ( so->scanFastResult == NULL )
! 				{
! 					/* scan of pending pages is finished */
! 					so->isScanFastInsert = false;
! 					tbm_end_iterate( so->scanFastIterator );
! 					so->scanFastIterator = NULL;
! 					startScan(scan);
! 					return scanGetItem(scan, item, recheck);
! 				}
! 				Assert( so->scanFastResult->ntuples >= 0 );
! 				so->scanFastOffset = 0;
! 			}
! 	
! 			ItemPointerSet(item,
! 							so->scanFastResult->blockno,
! 							so->scanFastResult->offsets[ so->scanFastOffset ]);
! 			*recheck = true; /* be conserative due to concurrent 
! 								removal from pending list */
! 
! 			so->scanFastOffset ++;
  
! 			return true;
! 		} 
! 		else
  		{
! 			/*
! 			 * Collect ItemPointers in memory
! 			 */
! 			so->scanFastTuples = scanFastInsert(scan, NULL, NULL);
  
! 			if ( so->scanFastTuples == NULL )
  			{
! 				/* nothing found */
! 				so->isScanFastInsert = false;
! 				startScan(scan);
  			}
  			else
! 			{
! 				so->scanFastIterator = tbm_begin_iterate(so->scanFastTuples);
! 			}
! 
! 			return scanGetItem(scan, item, recheck);
! 		}
! 	}
! 
! 	/*
! 	 * Regular scanning with filtering by already returned
! 	 * ItemPointers from pending list
! 	 */
! 
! 	do
! 	{
! 		ItemPointerSetMin(item);
! 		*recheck = false;
! 
! 		for (i = 0; i < so->nkeys; i++)
! 		{
! 			GinScanKey	key = so->keys + i;
! 
! 			if (keyGetItem(scan->indexRelation, &so->ginstate, so->tempCtx,
! 					   key, &keyrecheck))
! 				return FALSE;		/* finished one of keys */
! 			if (compareItemPointers(item, &key->curItem) < 0)
  				*item = key->curItem;
! 			*recheck |= keyrecheck;
! 		}
! 
! 		for (i = 1; i <= so->nkeys; i++)
! 		{
! 			GinScanKey	key = so->keys + i - 1;
! 
! 			for (;;)
! 			{
! 				int			cmp = compareItemPointers(item, &key->curItem);
! 	
! 				if (cmp == 0)
  				break;
+ 				else if (cmp > 0)
+ 				{
+ 					if (keyGetItem(scan->indexRelation, &so->ginstate, so->tempCtx,
+ 								   key, &keyrecheck))
+ 						return FALSE;		/* finished one of keys */
+ 					*recheck |= keyrecheck;
+ 				}
+ 				else
+ 				{					/* returns to begin */
+ 					*item = key->curItem;
+ 					i = 0;
+ 					break;
+ 				}
  			}
  		}
! 	} while( so->scanFastTuples && tbm_check_tuple(so->scanFastTuples, item) );
  
  	return TRUE;
  }
***************
*** 749,754 ****
--- 1169,1175 ----
  {
  	IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
  	TIDBitmap *tbm = (TIDBitmap *) PG_GETARG_POINTER(1);
+ 	GinScanOpaque so = (GinScanOpaque) scan->opaque;
  	int64		ntids;
  
  	if (GinIsNewKey(scan))
***************
*** 760,765 ****
--- 1181,1201 ----
  	startScan(scan);
  
  	ntids = 0;
+ 
+ 	/*
+ 	 * Small optimization: before regular scan of index let we execute 
+ 	 * scan of pending pages directly into result tbm. This saves 
+ 	 * cycles for filtering results and reduce memory usage. Also,
+ 	 * it allows to not check losiness of tbm
+ 	 */
+ 	scanFastInsert(scan, tbm, &ntids);
+ 
+ 	/*
+ 	 * Switch to regular scan
+ 	 */
+ 	so->isScanFastInsert = false;
+ 	startScan(scan);
+ 
  	for (;;)
  	{
  		ItemPointerData iptr;
diff -c -r -N src.orig/backend/access/gin/gininsert.c src/backend/access/gin/gininsert.c
*** src.orig/backend/access/gin/gininsert.c	2009-02-04 18:13:28.000000000 +0300
--- src/backend/access/gin/gininsert.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 138,144 ****
  /*
   * Inserts only one entry to the index, but it can add more than 1 ItemPointer.
   */
! static void
  ginEntryInsert(Relation index, GinState *ginstate, OffsetNumber attnum, Datum value, 
  				ItemPointerData *items, uint32 nitem, bool isBuild)
  {
--- 138,144 ----
  /*
   * Inserts only one entry to the index, but it can add more than 1 ItemPointer.
   */
! void
  ginEntryInsert(Relation index, GinState *ginstate, OffsetNumber attnum, Datum value, 
  				ItemPointerData *items, uint32 nitem, bool isBuild)
  {
***************
*** 273,279 ****
  	IndexBuildResult *result;
  	double		reltuples;
  	GinBuildState buildstate;
! 	Buffer		buffer;
  	ItemPointerData *list;
  	Datum		entry;
  	uint32		nlist;
--- 273,279 ----
  	IndexBuildResult *result;
  	double		reltuples;
  	GinBuildState buildstate;
! 	Buffer		RootBuffer, MetaBuffer;
  	ItemPointerData *list;
  	Datum		entry;
  	uint32		nlist;
***************
*** 286,296 ****
  
  	initGinState(&buildstate.ginstate, index);
  
  	/* initialize the root page */
! 	buffer = GinNewBuffer(index);
  	START_CRIT_SECTION();
! 	GinInitBuffer(buffer, GIN_LEAF);
! 	MarkBufferDirty(buffer);
  
  	if (!index->rd_istemp)
  	{
--- 286,302 ----
  
  	initGinState(&buildstate.ginstate, index);
  
+ 	/* initialize the meta page */
+ 	MetaBuffer = GinNewBuffer(index);
+ 
  	/* initialize the root page */
! 	RootBuffer = GinNewBuffer(index);
! 
  	START_CRIT_SECTION();
! 	GinInitMetabuffer(MetaBuffer);
! 	MarkBufferDirty(MetaBuffer);
! 	GinInitBuffer(RootBuffer, GIN_LEAF);
! 	MarkBufferDirty(RootBuffer);
  
  	if (!index->rd_istemp)
  	{
***************
*** 303,318 ****
  		rdata.len = sizeof(RelFileNode);
  		rdata.next = NULL;
  
- 		page = BufferGetPage(buffer);
- 
- 
  		recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_CREATE_INDEX, &rdata);
  		PageSetLSN(page, recptr);
  		PageSetTLI(page, ThisTimeLineID);
  
  	}
  
! 	UnlockReleaseBuffer(buffer);
  	END_CRIT_SECTION();
  
  	/* build the index */
--- 309,327 ----
  		rdata.len = sizeof(RelFileNode);
  		rdata.next = NULL;
  
  		recptr = XLogInsert(RM_GIN_ID, XLOG_GIN_CREATE_INDEX, &rdata);
+ 		
+ 		page = BufferGetPage(RootBuffer);
  		PageSetLSN(page, recptr);
  		PageSetTLI(page, ThisTimeLineID);
  
+ 		page = BufferGetPage(MetaBuffer);
+ 		PageSetLSN(page, recptr);
+ 		PageSetTLI(page, ThisTimeLineID);
  	}
  
! 	UnlockReleaseBuffer(MetaBuffer);
! 	UnlockReleaseBuffer(RootBuffer);
  	END_CRIT_SECTION();
  
  	/* build the index */
***************
*** 417,425 ****
  
  	initGinState(&ginstate, index);
  
! 	for(i=0; i<ginstate.origTupdesc->natts;i++)
! 		if ( !isnull[i] )
! 			res += ginHeapTupleInsert(index, &ginstate, (OffsetNumber)(i+1), values[i], ht_ctid);
  
  	MemoryContextSwitchTo(oldCtx);
  	MemoryContextDelete(insertCtx);
--- 426,451 ----
  
  	initGinState(&ginstate, index);
  
! 	if ( GinGetUseFastUpdate(index) )
! 	{
! 		GinTupleCollector	collector;
! 
! 		memset(&collector, 0, sizeof(GinTupleCollector));
! 		for(i=0; i<ginstate.origTupdesc->natts;i++)
! 			if ( !isnull[i] )
! 				res += ginHeapTupleFastCollect(index, &ginstate, &collector,
! 												(OffsetNumber)(i+1), values[i], ht_ctid);
! 
! 		ginHeapTupleFastInsert(index, &ginstate, &collector);
! 	}
! 	else
! 	{
! 		for(i=0; i<ginstate.origTupdesc->natts;i++)
! 			if ( !isnull[i] ) 
! 				res += ginHeapTupleInsert(index, &ginstate, 
! 												(OffsetNumber)(i+1), values[i], ht_ctid);
! 
! 	}
  
  	MemoryContextSwitchTo(oldCtx);
  	MemoryContextDelete(insertCtx);
diff -c -r -N src.orig/backend/access/gin/ginscan.c src/backend/access/gin/ginscan.c
*** src.orig/backend/access/gin/ginscan.c	2009-02-04 18:13:28.000000000 +0300
--- src/backend/access/gin/ginscan.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 164,169 ****
--- 164,173 ----
  				 errmsg("GIN indexes do not support whole-index scans")));
  
  	so->isVoidRes = false;
+ 	so->isScanFastInsert = true;
+ 	so->scanFastTuples = NULL;
+ 	so->scanFastResult = NULL;
+ 	so->scanFastIterator = NULL;
  
  	for (i = 0; i < scan->numberOfKeys; i++)
  	{
***************
*** 238,245 ****
--- 242,255 ----
  	else
  	{
  		freeScanKeys(so->keys, so->nkeys);
+ 		if ( so->scanFastIterator )
+ 			tbm_end_iterate( so->scanFastIterator );
+ 		if ( so->scanFastTuples )
+ 			tbm_free( so->scanFastTuples );
  	}
  
+ 	so->scanFastIterator = NULL;
+ 	so->scanFastTuples = NULL;
  	so->keys = NULL;
  
  	if (scankey && scan->numberOfKeys > 0)
***************
*** 261,266 ****
--- 271,280 ----
  	if (so != NULL)
  	{
  		freeScanKeys(so->keys, so->nkeys);
+ 		if ( so->scanFastIterator )
+ 			tbm_end_iterate( so->scanFastIterator );
+ 		if ( so->scanFastTuples )
+ 			tbm_free( so->scanFastTuples );
  
  		MemoryContextDelete(so->tempCtx);
  
diff -c -r -N src.orig/backend/access/gin/ginutil.c src/backend/access/gin/ginutil.c
*** src.orig/backend/access/gin/ginutil.c	2009-02-04 18:13:28.000000000 +0300
--- src/backend/access/gin/ginutil.c	2009-02-04 18:44:12.000000000 +0300
***************
*** 21,26 ****
--- 21,27 ----
  #include "storage/freespace.h"
  #include "storage/indexfsm.h"
  #include "storage/lmgr.h"
+ #include "utils/guc.h"
  
  void
  initGinState(GinState *state, Relation index)
***************
*** 57,63 ****
  						CurrentMemoryContext);
  
  		/*
! 		 * Check opclass capability to do partial match. 
  		 */
  		if ( index_getprocid(index, i+1, GIN_COMPARE_PARTIAL_PROC) != InvalidOid )
  		{
--- 58,64 ----
  						CurrentMemoryContext);
  
  		/*
! 		 * Check opclass capability to do partial match.
  		 */
  		if ( index_getprocid(index, i+1, GIN_COMPARE_PARTIAL_PROC) != InvalidOid )
  		{
***************
*** 88,94 ****
  		bool    isnull;
  
  		/*
! 		 * First attribute is always int16, so we can safely use any 
  		 * tuple descriptor to obtain first attribute of tuple
  		 */
  		res = index_getattr(tuple, FirstOffsetNumber, ginstate->tupdesc[0],
--- 89,95 ----
  		bool    isnull;
  
  		/*
! 		 * First attribute is always int16, so we can safely use any
  		 * tuple descriptor to obtain first attribute of tuple
  		 */
  		res = index_getattr(tuple, FirstOffsetNumber, ginstate->tupdesc[0],
***************
*** 213,218 ****
--- 214,235 ----
  	GinInitPage(BufferGetPage(b), f, BufferGetPageSize(b));
  }
  
+ void
+ GinInitMetabuffer(Buffer b)
+ {
+ 	GinMetaPageData	*metadata;
+ 	Page 			 page = BufferGetPage(b);
+ 
+ 	GinInitPage(page, GIN_META, BufferGetPageSize(b));
+ 
+ 	metadata = GinPageGetMeta(page);
+ 
+ 	metadata->head = metadata->tail = InvalidBlockNumber;
+ 	metadata->tailFreeSize = 0;
+ 	metadata->nPendingPages = 0;
+ 	metadata->nPendingHeapTuples = 0;
+ }
+ 
  int
  compareEntries(GinState *ginstate, OffsetNumber attnum, Datum a, Datum b)
  {
***************
*** 313,324 ****
  Datum
  ginoptions(PG_FUNCTION_ARGS)
  {
! 	Datum		reloptions = PG_GETARG_DATUM(0);
! 	bool		validate = PG_GETARG_BOOL(1);
! 	bytea	   *result;
! 
! 	result = default_reloptions(reloptions, validate, RELOPT_KIND_GIN);
! 	if (result)
! 		PG_RETURN_BYTEA_P(result);
! 	PG_RETURN_NULL();
  }
--- 330,357 ----
  Datum
  ginoptions(PG_FUNCTION_ARGS)
  {
! 	Datum       reloptions = PG_GETARG_DATUM(0);
! 	bool        validate = PG_GETARG_BOOL(1);
! 	relopt_value   *options;
! 	GinOptions     *rdopts;
! 	int             numoptions;
! 	relopt_parse_elt tab[] = {
! 		{"fastupdate", RELOPT_TYPE_BOOL, offsetof(GinOptions, useFastUpdate)}
! 	};
! 
! 	options = parseRelOptions(reloptions, validate, RELOPT_KIND_GIN, &numoptions);
! 
! 	/* if none set, we're done */
! 	if (numoptions == 0)
! 		PG_RETURN_NULL();
! 
! 	rdopts = allocateReloptStruct(sizeof(GinOptions), options, numoptions);
! 
! 	fillRelOptions((void *) rdopts, sizeof(GinOptions), options, numoptions,
! 					validate, tab, lengthof(tab));
! 																				
! 	pfree(options);
! 																					  
! 	PG_RETURN_BYTEA_P(rdopts);
  }
+ 
diff -c -r -N src.orig/backend/access/gin/ginvacuum.c src/backend/access/gin/ginvacuum.c
*** src.orig/backend/access/gin/ginvacuum.c	2009-02-04 18:13:28.000000000 +0300
--- src/backend/access/gin/ginvacuum.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 593,610 ****
  	BlockNumber rootOfPostingTree[BLCKSZ / (sizeof(IndexTupleData) + sizeof(ItemId))];
  	uint32		nRoot;
  
  	/* first time through? */
  	if (stats == NULL)
  		stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
  	/* we'll re-count the tuples each time */
  	stats->num_index_tuples = 0;
- 
- 	gvs.index = index;
  	gvs.result = stats;
- 	gvs.callback = callback;
- 	gvs.callback_state = callback_state;
- 	gvs.strategy = info->strategy;
- 	initGinState(&gvs.ginstate, index);
  
  	buffer = ReadBufferExtended(index, MAIN_FORKNUM, blkno,
  								RBM_NORMAL, info->strategy);
--- 593,614 ----
  	BlockNumber rootOfPostingTree[BLCKSZ / (sizeof(IndexTupleData) + sizeof(ItemId))];
  	uint32		nRoot;
  
+ 	gvs.index = index;
+ 	gvs.callback = callback;
+ 	gvs.callback_state = callback_state;
+ 	gvs.strategy = info->strategy;
+ 	initGinState(&gvs.ginstate, index);
+ 
  	/* first time through? */
  	if (stats == NULL)
+ 	{
  		stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
+ 		ginInsertCleanup(index, &gvs.ginstate, stats);
+ 	}
+ 
  	/* we'll re-count the tuples each time */
  	stats->num_index_tuples = 0;
  	gvs.result = stats;
  
  	buffer = ReadBufferExtended(index, MAIN_FORKNUM, blkno,
  								RBM_NORMAL, info->strategy);
***************
*** 703,711 ****
  	BlockNumber lastBlock = GIN_ROOT_BLKNO,
  				lastFilledBlock = GIN_ROOT_BLKNO;
  
! 	/* Set up all-zero stats if ginbulkdelete wasn't called */
  	if (stats == NULL)
  		stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
  
  	/*
  	 * XXX we always report the heap tuple count as the number of index
--- 707,724 ----
  	BlockNumber lastBlock = GIN_ROOT_BLKNO,
  				lastFilledBlock = GIN_ROOT_BLKNO;
  
! 	/* 
! 	 * Set up all-zero stats and finalyze fast insertion 
! 	 * if ginbulkdelete wasn't called 
! 	 */
  	if (stats == NULL)
+ 	{
+ 		GinState	ginstate;
+ 
+ 		initGinState(&ginstate, index);
  		stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
+ 		ginInsertCleanup(index, &ginstate, stats);
+ 	}
  
  	/*
  	 * XXX we always report the heap tuple count as the number of index
diff -c -r -N src.orig/backend/access/gin/ginxlog.c src/backend/access/gin/ginxlog.c
*** src.orig/backend/access/gin/ginxlog.c	2009-02-04 18:13:28.000000000 +0300
--- src/backend/access/gin/ginxlog.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 71,90 ****
  ginRedoCreateIndex(XLogRecPtr lsn, XLogRecord *record)
  {
  	RelFileNode *node = (RelFileNode *) XLogRecGetData(record);
! 	Buffer		buffer;
  	Page		page;
  
! 	buffer = XLogReadBuffer(*node, GIN_ROOT_BLKNO, true);
! 	Assert(BufferIsValid(buffer));
! 	page = (Page) BufferGetPage(buffer);
  
! 	GinInitBuffer(buffer, GIN_LEAF);
  
  	PageSetLSN(page, lsn);
  	PageSetTLI(page, ThisTimeLineID);
  
! 	MarkBufferDirty(buffer);
! 	UnlockReleaseBuffer(buffer);
  }
  
  static void
--- 71,100 ----
  ginRedoCreateIndex(XLogRecPtr lsn, XLogRecord *record)
  {
  	RelFileNode *node = (RelFileNode *) XLogRecGetData(record);
! 	Buffer		RootBuffer, MetaBuffer;
  	Page		page;
  
! 	MetaBuffer = XLogReadBuffer(*node, GIN_METAPAGE_BLKNO, true);
! 	Assert(BufferIsValid(MetaBuffer));
! 	GinInitMetabuffer(MetaBuffer);
! 
! 	page = (Page) BufferGetPage(MetaBuffer);
! 	PageSetLSN(page, lsn);
! 	PageSetTLI(page, ThisTimeLineID);
  
! 	RootBuffer = XLogReadBuffer(*node, GIN_ROOT_BLKNO, true);
! 	Assert(BufferIsValid(RootBuffer));
! 	page = (Page) BufferGetPage(RootBuffer);
! 
! 	GinInitBuffer(RootBuffer, GIN_LEAF);
  
  	PageSetLSN(page, lsn);
  	PageSetTLI(page, ThisTimeLineID);
  
! 	MarkBufferDirty(MetaBuffer);
! 	UnlockReleaseBuffer(MetaBuffer);
! 	MarkBufferDirty(RootBuffer);
! 	UnlockReleaseBuffer(RootBuffer);
  }
  
  static void
***************
*** 433,438 ****
--- 443,616 ----
  	}
  }
  
+ static void
+ ginRedoUpdateMetapage(XLogRecPtr lsn, XLogRecord *record)
+ {
+ 	ginxlogUpdateMeta *data = (ginxlogUpdateMeta*) XLogRecGetData(record);
+ 	Buffer		metabuffer;
+ 	Page		metapage;
+ 
+ 	metabuffer = XLogReadBuffer(data->node, GIN_METAPAGE_BLKNO, false);
+ 	metapage = BufferGetPage(metabuffer);
+ 
+ 	if (!XLByteLE(lsn, PageGetLSN(metapage)))
+ 	{
+ 		memcpy( GinPageGetMeta(metapage), &data->metadata, sizeof(GinMetaPageData));
+ 		PageSetLSN(metapage, lsn);
+ 		PageSetTLI(metapage, ThisTimeLineID);
+ 		MarkBufferDirty(metabuffer);
+ 	}
+ 
+ 	if ( data->ntuples > 0 )
+ 	{
+ 		/*
+ 		 * insert into tail page
+ 		 */
+ 		if (!(record->xl_info & XLR_BKP_BLOCK_1))
+ 		{
+ 			Buffer 	buffer = XLogReadBuffer(data->node, data->metadata.tail, false);
+ 			Page 	page = BufferGetPage(buffer);
+ 
+ 			if ( !XLByteLE(lsn, PageGetLSN(page)))
+ 			{
+ 				OffsetNumber l, off = (PageIsEmpty(page)) ? FirstOffsetNumber :
+ 						OffsetNumberNext(PageGetMaxOffsetNumber(page));
+ 				int				i, tupsize;
+ 				IndexTuple		tuples = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogUpdateMeta));
+ 
+ 				for(i=0; i<data->ntuples; i++)
+ 				{
+ 					tupsize = IndexTupleSize(tuples);
+ 
+ 					l = PageAddItem(page, (Item)tuples, tupsize, off, false, false);
+ 
+ 					if (l == InvalidOffsetNumber)
+ 						elog(ERROR, "failed to add item to index page");
+ 
+ 					tuples = (IndexTuple)( ((char*)tuples) + tupsize );
+ 				}
+ 
+ 				/*
+ 				 * Increase counter of heap tuples
+ 				 */
+ 				GinPageGetOpaque(page)->maxoff++;
+ 
+ 				PageSetLSN(page, lsn);
+ 				PageSetTLI(page, ThisTimeLineID);
+ 				MarkBufferDirty(buffer);
+ 			}
+ 			UnlockReleaseBuffer(buffer);
+ 		}
+ 	}
+ 	else if ( data->prevTail != InvalidBlockNumber )
+ 	{
+ 		/*
+ 		 * New tail
+ 		 */
+ 
+ 		Buffer 	buffer = XLogReadBuffer(data->node, data->prevTail, false);
+ 		Page 	page = BufferGetPage(buffer);
+ 
+ 		if ( !XLByteLE(lsn, PageGetLSN(page)))
+ 		{
+ 			GinPageGetOpaque(page)->rightlink = data->newRightlink;
+ 
+ 			PageSetLSN(page, lsn);
+ 			PageSetTLI(page, ThisTimeLineID);
+ 			MarkBufferDirty(buffer);
+ 		}
+ 		UnlockReleaseBuffer(buffer);
+ 	}
+ 
+ 	UnlockReleaseBuffer(metabuffer);
+ }
+ 
+ static void
+ ginRedoInsertListPage(XLogRecPtr lsn, XLogRecord *record)
+ {
+ 	ginxlogInsertListPage *data = (ginxlogInsertListPage*) XLogRecGetData(record);
+ 	Buffer			buffer;
+ 	Page			page;
+ 	OffsetNumber 	l, off = FirstOffsetNumber;
+ 	int				i, tupsize;
+ 	IndexTuple      tuples = (IndexTuple) (XLogRecGetData(record) + sizeof(ginxlogInsertListPage));
+ 
+ 	if (record->xl_info & XLR_BKP_BLOCK_1)
+ 		return;
+ 
+ 	buffer = XLogReadBuffer(data->node, data->blkno, true);
+ 	page = BufferGetPage(buffer);
+ 
+ 	GinInitBuffer(buffer, GIN_LIST);
+ 	GinPageGetOpaque(page)->rightlink = data->rightlink;
+ 	if ( data->rightlink == InvalidBlockNumber )
+ 	{
+ 		/* tail of sublist */
+ 		GinPageSetFullRow(page);
+ 		GinPageGetOpaque(page)->maxoff = 1;
+ 	}
+ 	else
+ 	{
+ 		GinPageGetOpaque(page)->maxoff = 0;
+ 	}
+ 
+ 	for(i=0; i<data->ntuples; i++)
+ 	{
+ 		tupsize = IndexTupleSize(tuples);
+ 
+ 		l = PageAddItem(page, (Item)tuples, tupsize, off, false, false);
+ 
+ 		if (l == InvalidOffsetNumber)
+ 			elog(ERROR, "failed to add item to index page");
+ 
+ 		tuples = (IndexTuple)( ((char*)tuples) + tupsize );
+ 	}
+ 
+ 	PageSetLSN(page, lsn);
+ 	PageSetTLI(page, ThisTimeLineID);
+ 	MarkBufferDirty(buffer);
+ 
+ 	UnlockReleaseBuffer(buffer);
+ }
+ 
+ static void
+ ginRedoDeleteListPages(XLogRecPtr lsn, XLogRecord *record)
+ {
+ 	ginxlogDeleteListPages *data = (ginxlogDeleteListPages*) XLogRecGetData(record);
+ 	Buffer		metabuffer;
+ 	Page		metapage;
+ 	int 		i;
+ 
+ 	metabuffer = XLogReadBuffer(data->node, GIN_METAPAGE_BLKNO, false);
+ 	metapage = BufferGetPage(metabuffer);
+ 
+ 	if (!XLByteLE(lsn, PageGetLSN(metapage)))
+ 	{
+ 		memcpy( GinPageGetMeta(metapage), &data->metadata, sizeof(GinMetaPageData));
+ 		PageSetLSN(metapage, lsn);
+ 		PageSetTLI(metapage, ThisTimeLineID);
+ 		MarkBufferDirty(metabuffer);
+ 	}
+ 
+ 	for(i=0; i<data->ndeleted; i++)
+ 	{
+ 		Buffer 	buffer = XLogReadBuffer(data->node,data->toDelete[i],false);
+ 		Page	page = BufferGetPage(buffer);
+ 
+ 		if ( !XLByteLE(lsn, PageGetLSN(page)))
+ 		{
+ 			GinPageGetOpaque(page)->flags = GIN_DELETED;
+ 
+ 			PageSetLSN(page, lsn);
+ 			PageSetTLI(page, ThisTimeLineID);
+ 			MarkBufferDirty(buffer);
+ 		}
+ 
+ 		UnlockReleaseBuffer(buffer);
+ 	}
+ 	UnlockReleaseBuffer(metabuffer);
+ }
+ 
  void
  gin_redo(XLogRecPtr lsn, XLogRecord *record)
  {
***************
*** 461,466 ****
--- 639,653 ----
  		case XLOG_GIN_DELETE_PAGE:
  			ginRedoDeletePage(lsn, record);
  			break;
+ 		case XLOG_GIN_UPDATE_META_PAGE:
+ 			ginRedoUpdateMetapage(lsn, record);
+ 			break;
+ 		case XLOG_GIN_INSERT_LISTPAGE:
+ 			ginRedoInsertListPage(lsn, record);
+ 			break;
+ 		case XLOG_GIN_DELETE_LISTPAGE: 
+ 			ginRedoDeleteListPages(lsn, record);
+ 			break;
  		default:
  			elog(PANIC, "gin_redo: unknown op code %u", info);
  	}
***************
*** 516,521 ****
--- 703,720 ----
  			appendStringInfo(buf, "Delete page, ");
  			desc_node(buf, ((ginxlogDeletePage *) rec)->node, ((ginxlogDeletePage *) rec)->blkno);
  			break;
+ 		case XLOG_GIN_UPDATE_META_PAGE:
+ 			appendStringInfo(buf, "Update metapage, ");
+ 			desc_node(buf, ((ginxlogUpdateMeta *) rec)->node, ((ginxlogUpdateMeta *) rec)->metadata.tail); 
+ 			break;
+ 		case XLOG_GIN_INSERT_LISTPAGE:
+ 			appendStringInfo(buf, "insert new list page, ");
+ 			desc_node(buf, ((ginxlogInsertListPage *) rec)->node, ((ginxlogInsertListPage *) rec)->blkno); 
+ 			break;
+ 		case XLOG_GIN_DELETE_LISTPAGE:
+ 			appendStringInfo(buf, "Delete list page (%d), ", ((ginxlogDeleteListPages *) rec)->ndeleted);
+ 			desc_node(buf, ((ginxlogDeleteListPages *) rec)->node, ((ginxlogDeleteListPages *) rec)->metadata.head); 
+ 			break;
  		default:
  			elog(PANIC, "gin_desc: unknown op code %u", info);
  	}
diff -c -r -N src.orig/backend/catalog/system_views.sql src/backend/catalog/system_views.sql
*** src.orig/backend/catalog/system_views.sql	2009-02-04 18:13:28.000000000 +0300
--- src/backend/catalog/system_views.sql	2009-02-04 18:23:22.000000000 +0300
***************
*** 193,198 ****
--- 193,199 ----
              pg_stat_get_tuples_updated(C.oid) AS n_tup_upd, 
              pg_stat_get_tuples_deleted(C.oid) AS n_tup_del,
              pg_stat_get_tuples_hot_updated(C.oid) AS n_tup_hot_upd,
+             pg_stat_get_fresh_inserted_tuples(C.oid) AS n_fresh_tup, 
              pg_stat_get_live_tuples(C.oid) AS n_live_tup, 
              pg_stat_get_dead_tuples(C.oid) AS n_dead_tup,
              pg_stat_get_last_vacuum_time(C.oid) as last_vacuum,
diff -c -r -N src.orig/backend/nodes/tidbitmap.c src/backend/nodes/tidbitmap.c
*** src.orig/backend/nodes/tidbitmap.c	2009-02-04 18:13:29.000000000 +0300
--- src/backend/nodes/tidbitmap.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 169,174 ****
--- 169,195 ----
  static void tbm_lossify(TIDBitmap *tbm);
  static int	tbm_comparator(const void *left, const void *right);
  
+ /*
+  * Estimate guaranteed number of tuples which can be stored in 
+  * tidbitmap without lossify. This estimates the hash overhead 
+  * at MAXALIGN(sizeof(HASHELEMENT)) plus a pointer per hash entry, 
+  * which is crude but good enough for our purpose.
+  * Also count an extra Pointer per entry for the arrays created during
+  * iteration readout.
+  */
+ long 
+ tbm_max_non_lossy(long maxbytes)
+ {
+ 	long result;
+ 
+ 	result = maxbytes /
+ 		(MAXALIGN(sizeof(HASHELEMENT)) + MAXALIGN(sizeof(PagetableEntry))
+ 		 + sizeof(Pointer) + sizeof(Pointer));
+ 	result = Min(result, INT_MAX - 1);		/* safety limit */
+ 	result = Max(result, 16);		/* sanity limit */
+ 
+ 	return result;
+ }
  
  /*
   * tbm_create - create an initially-empty bitmap
***************
*** 181,187 ****
  tbm_create(long maxbytes)
  {
  	TIDBitmap  *tbm;
- 	long		nbuckets;
  
  	/* Create the TIDBitmap struct and zero all its fields */
  	tbm = makeNode(TIDBitmap);
--- 202,207 ----
***************
*** 190,207 ****
  	tbm->status = TBM_EMPTY;
  
  	/*
! 	 * Estimate number of hashtable entries we can have within maxbytes. This
! 	 * estimates the hash overhead at MAXALIGN(sizeof(HASHELEMENT)) plus a
! 	 * pointer per hash entry, which is crude but good enough for our purpose.
! 	 * Also count an extra Pointer per entry for the arrays created during
! 	 * iteration readout.
  	 */
! 	nbuckets = maxbytes /
! 		(MAXALIGN(sizeof(HASHELEMENT)) + MAXALIGN(sizeof(PagetableEntry))
! 		 + sizeof(Pointer) + sizeof(Pointer));
! 	nbuckets = Min(nbuckets, INT_MAX - 1);		/* safety limit */
! 	nbuckets = Max(nbuckets, 16);		/* sanity limit */
! 	tbm->maxentries = (int) nbuckets;
  
  	return tbm;
  }
--- 210,218 ----
  	tbm->status = TBM_EMPTY;
  
  	/*
! 	 * Estimate number of hashtable entries we can have within maxbytes.
  	 */
! 	tbm->maxentries = (int) tbm_max_non_lossy(maxbytes);
  
  	return tbm;
  }
***************
*** 310,315 ****
--- 321,367 ----
  }
  
  /*
+  * tbm_check_tuple - Check presence of tuple's ID in a TIDBitmap
+  */
+ bool
+ tbm_check_tuple(TIDBitmap *tbm, const ItemPointer tid) {
+ 	BlockNumber 	blk = ItemPointerGetBlockNumber(tid);
+ 	OffsetNumber 	off = ItemPointerGetOffsetNumber(tid);
+ 	PagetableEntry *page;
+ 	int				wordnum,
+ 					bitnum;
+ 
+ 	/* safety check to ensure we don't overrun bit array bounds */
+ 	if (off < 1 || off > MAX_TUPLES_PER_PAGE)
+ 		elog(ERROR, "tuple offset out of range: %u", off);
+ 
+ 	if (tbm_page_is_lossy(tbm, blk))
+ 		return true; /* whole page is already marked */
+ 
+ 	page = tbm_get_pageentry(tbm, blk);
+ 	if (page->ischunk)
+ 	{
+ 		wordnum = bitnum = 0;
+ 	}
+ 	else
+ 	{
+ 		wordnum = WORDNUM(off - 1);
+ 		bitnum = BITNUM(off - 1);
+ 	}
+ 
+ 	return ( page->words[wordnum] & ((bitmapword) 1 << bitnum) ) ? true : false; 
+ }
+ 
+ /*
+  * tbm_has_lossy - returns true if there is at least one lossy page
+  */
+ bool
+ tbm_has_lossy(TIDBitmap *tbm)
+ {
+ 	return (tbm->nchunks>0);
+ }
+ 
+ /*
   * tbm_union - set union
   *
   * a is modified in-place, b is not changed
diff -c -r -N src.orig/backend/postmaster/autovacuum.c src/backend/postmaster/autovacuum.c
*** src.orig/backend/postmaster/autovacuum.c	2009-02-04 18:13:30.000000000 +0300
--- src/backend/postmaster/autovacuum.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 71,76 ****
--- 71,77 ----
  #include "access/heapam.h"
  #include "access/transam.h"
  #include "access/xact.h"
+ #include "access/gin.h"
  #include "catalog/dependency.h"
  #include "catalog/indexing.h"
  #include "catalog/namespace.h"
***************
*** 2492,2497 ****
--- 2493,2550 ----
  }
  
  /*
+  * relation_has_pending_indexes
+  *
+  * Returns true if relation has indexes with delayed insertion.
+  * Currently, only GIN has that possiblity
+  */
+ 
+ static bool
+ relation_has_pending_indexes(Oid relid, Form_pg_class classForm)
+ {
+ 	Relation	rel;
+ 	List       *indexoidlist;
+ 	ListCell   *indexoidscan;
+ 	bool		has = false;
+ 
+ 	/* only ordinary cataloged heap can contains such indexes */
+ 	if ( classForm->relkind != RELKIND_RELATION )
+ 		return false;
+ 
+ 	/* has not indexes at all */
+ 	if ( classForm->relhasindex == false )
+ 		return false;
+ 
+ 	rel = RelationIdGetRelation(relid);
+ 
+ 	indexoidlist = RelationGetIndexList(rel);
+ 
+ 	foreach(indexoidscan, indexoidlist)
+ 	{
+ 		Oid         indexoid = lfirst_oid(indexoidscan);
+ 		Relation	irel = RelationIdGetRelation(indexoid);
+ 
+ 		/*
+ 		 *  Currently, only GIN in fast update mode  
+ 		 */
+ 		if ( irel->rd_rel->relam == GIN_AM_OID && GinGetUseFastUpdate(irel) )
+ 		{
+ 			RelationClose(irel);
+ 			has = true;
+ 			break;
+ 		}
+ 
+ 		RelationClose(irel);
+ 	}
+ 
+ 	list_free(indexoidlist);
+ 
+ 	RelationClose(rel);
+ 
+ 	return has;
+ }
+ 
+ /*
   * relation_needs_vacanalyze
   *
   * Check whether a relation needs to be vacuumed or analyzed; return each into
***************
*** 2548,2554 ****
  
  	/* number of vacuum (resp. analyze) tuples at this time */
  	float4		vactuples,
! 				anltuples;
  
  	/* freeze parameters */
  	int			freeze_max_age;
--- 2601,2608 ----
  
  	/* number of vacuum (resp. analyze) tuples at this time */
  	float4		vactuples,
! 				anltuples,
! 				instuples;
  
  	/* freeze parameters */
  	int			freeze_max_age;
***************
*** 2613,2618 ****
--- 2667,2673 ----
  		vactuples = tabentry->n_dead_tuples;
  		anltuples = tabentry->n_live_tuples + tabentry->n_dead_tuples -
  			tabentry->last_anl_tuples;
+ 		instuples = tabentry->n_inserted_tuples;
  
  		vacthresh = (float4) vac_base_thresh + vac_scale_factor * reltuples;
  		anlthresh = (float4) anl_base_thresh + anl_scale_factor * reltuples;
***************
*** 2626,2633 ****
  			 NameStr(classForm->relname),
  			 vactuples, vacthresh, anltuples, anlthresh);
  
! 		/* Determine if this table needs vacuum or analyze. */
! 		*dovacuum = force_vacuum || (vactuples > vacthresh);
  		*doanalyze = (anltuples > anlthresh);
  	}
  	else
--- 2681,2693 ----
  			 NameStr(classForm->relname),
  			 vactuples, vacthresh, anltuples, anlthresh);
  
! 		/* 
! 		 * Determine if this table needs vacuum or analyze. 
! 		 * Use vac_base_thresh as a theshhold for instuples because
! 		 * search time of GIN's pending pages is linear by its number. 
! 		 */
! 		*dovacuum = force_vacuum || (vactuples > vacthresh) || 
! 				(relation_has_pending_indexes(relid, classForm) && instuples > vac_base_thresh);
  		*doanalyze = (anltuples > anlthresh);
  	}
  	else
diff -c -r -N src.orig/backend/postmaster/pgstat.c src/backend/postmaster/pgstat.c
*** src.orig/backend/postmaster/pgstat.c	2009-02-04 18:13:30.000000000 +0300
--- src/backend/postmaster/pgstat.c	2009-02-04 18:23:22.000000000 +0300
***************
*** 3537,3542 ****
--- 3537,3545 ----
  			tabentry->tuples_updated = tabmsg[i].t_counts.t_tuples_updated;
  			tabentry->tuples_deleted = tabmsg[i].t_counts.t_tuples_deleted;
  			tabentry->tuples_hot_updated = tabmsg[i].t_counts.t_tuples_hot_updated;
+ 			tabentry->n_inserted_tuples = tabmsg[i].t_counts.t_tuples_inserted + 
+ 										  tabmsg[i].t_counts.t_tuples_updated -
+ 										  tabmsg[i].t_counts.t_tuples_hot_updated;
  			tabentry->n_live_tuples = tabmsg[i].t_counts.t_new_live_tuples;
  			tabentry->n_dead_tuples = tabmsg[i].t_counts.t_new_dead_tuples;
  			tabentry->blocks_fetched = tabmsg[i].t_counts.t_blocks_fetched;
***************
*** 3560,3565 ****
--- 3563,3571 ----
  			tabentry->tuples_updated += tabmsg[i].t_counts.t_tuples_updated;
  			tabentry->tuples_deleted += tabmsg[i].t_counts.t_tuples_deleted;
  			tabentry->tuples_hot_updated += tabmsg[i].t_counts.t_tuples_hot_updated;
+ 			tabentry->n_inserted_tuples += tabmsg[i].t_counts.t_tuples_inserted + 
+ 										  tabmsg[i].t_counts.t_tuples_updated -
+ 										  tabmsg[i].t_counts.t_tuples_hot_updated;
  			tabentry->n_live_tuples += tabmsg[i].t_counts.t_new_live_tuples;
  			tabentry->n_dead_tuples += tabmsg[i].t_counts.t_new_dead_tuples;
  			tabentry->blocks_fetched += tabmsg[i].t_counts.t_blocks_fetched;
***************
*** 3570,3575 ****
--- 3576,3583 ----
  		tabentry->n_live_tuples = Max(tabentry->n_live_tuples, 0);
  		/* Likewise for n_dead_tuples */
  		tabentry->n_dead_tuples = Max(tabentry->n_dead_tuples, 0);
+ 		/* Likewise for n_inserted_tuples */
+ 		tabentry->n_inserted_tuples = Max(tabentry->n_inserted_tuples, 0);
  
  		/*
  		 * Add per-table stats to the per-database entry, too.
***************
*** 3770,3775 ****
--- 3778,3784 ----
  		tabentry->n_live_tuples = msg->m_tuples;
  	/* Resetting dead_tuples to 0 is an approximation ... */
  	tabentry->n_dead_tuples = 0;
+ 	tabentry->n_inserted_tuples = 0;
  	if (msg->m_analyze)
  	{
  		if (msg->m_scanned_all)
diff -c -r -N src.orig/backend/utils/adt/pgstatfuncs.c src/backend/utils/adt/pgstatfuncs.c
*** src.orig/backend/utils/adt/pgstatfuncs.c	2009-02-04 18:13:32.000000000 +0300
--- src/backend/utils/adt/pgstatfuncs.c	2009-02-04 18:45:50.000000000 +0300
***************
*** 209,214 ****
--- 209,228 ----
  	PG_RETURN_INT64(result);
  }
  
+ Datum
+ pg_stat_get_fresh_inserted_tuples(PG_FUNCTION_ARGS)
+ {
+ 	Oid			relid = PG_GETARG_OID(0);
+ 	int64		result;
+ 	PgStat_StatTabEntry *tabentry;
+ 
+ 	if ((tabentry = pgstat_fetch_stat_tabentry(relid)) == NULL)
+ 		result = 0;
+ 	else
+ 		result = (int64) (tabentry->n_inserted_tuples);
+ 
+ 	PG_RETURN_INT64(result);
+ }
  
  Datum
  pg_stat_get_dead_tuples(PG_FUNCTION_ARGS)
diff -c -r -N src.orig/backend/utils/adt/selfuncs.c src/backend/utils/adt/selfuncs.c
*** src.orig/backend/utils/adt/selfuncs.c	2009-02-04 18:13:32.000000000 +0300
--- src/backend/utils/adt/selfuncs.c	2009-02-04 19:47:19.000000000 +0300
***************
*** 92,103 ****
--- 92,106 ----
  #include <math.h>
  
  #include "access/sysattr.h"
+ #include "access/gin.h"
  #include "catalog/pg_opfamily.h"
  #include "catalog/pg_statistic.h"
  #include "catalog/pg_type.h"
+ #include "miscadmin.h"
  #include "mb/pg_wchar.h"
  #include "nodes/makefuncs.h"
  #include "nodes/nodeFuncs.h"
+ #include "nodes/tidbitmap.h"
  #include "optimizer/clauses.h"
  #include "optimizer/cost.h"
  #include "optimizer/pathnode.h"
***************
*** 108,113 ****
--- 111,117 ----
  #include "optimizer/var.h"
  #include "parser/parse_coerce.h"
  #include "parser/parsetree.h"
+ #include "pgstat.h"
  #include "utils/builtins.h"
  #include "utils/date.h"
  #include "utils/datum.h"
***************
*** 5940,5957 ****
  Datum
  gincostestimate(PG_FUNCTION_ARGS)
  {
! 	PlannerInfo *root = (PlannerInfo *) PG_GETARG_POINTER(0);
! 	IndexOptInfo *index = (IndexOptInfo *) PG_GETARG_POINTER(1);
! 	List	   *indexQuals = (List *) PG_GETARG_POINTER(2);
! 	RelOptInfo *outer_rel = (RelOptInfo *) PG_GETARG_POINTER(3);
! 	Cost	   *indexStartupCost = (Cost *) PG_GETARG_POINTER(4);
! 	Cost	   *indexTotalCost = (Cost *) PG_GETARG_POINTER(5);
! 	Selectivity *indexSelectivity = (Selectivity *) PG_GETARG_POINTER(6);
! 	double	   *indexCorrelation = (double *) PG_GETARG_POINTER(7);
  
  	genericcostestimate(root, index, indexQuals, outer_rel, 0.0,
  						indexStartupCost, indexTotalCost,
  						indexSelectivity, indexCorrelation);
  
  	PG_RETURN_VOID();
  }
--- 5944,6050 ----
  Datum
  gincostestimate(PG_FUNCTION_ARGS)
  {
! 	PlannerInfo    *root = (PlannerInfo *) PG_GETARG_POINTER(0);
! 	IndexOptInfo   *index = (IndexOptInfo *) PG_GETARG_POINTER(1);
! 	List		   *indexQuals = (List *) PG_GETARG_POINTER(2);
! 	RelOptInfo 	   *outer_rel = (RelOptInfo *) PG_GETARG_POINTER(3);
! 	Cost		   *indexStartupCost = (Cost *) PG_GETARG_POINTER(4);
! 	Cost		   *indexTotalCost = (Cost *) PG_GETARG_POINTER(5);
! 	Selectivity	   *indexSelectivity = (Selectivity *) PG_GETARG_POINTER(6);
! 	double		   *indexCorrelation = (double *) PG_GETARG_POINTER(7);
! 	int64			n_pending_tuples;
! 	Cost			pendingPageCost;
! 	RangeTblEntry	*rte;
! 	Relation		irel;
  
  	genericcostestimate(root, index, indexQuals, outer_rel, 0.0,
  						indexStartupCost, indexTotalCost,
  						indexSelectivity, indexCorrelation);
  
+ 	/*
+ 	 * Check index mode
+ 	 */
+ 	irel = RelationIdGetRelation( index->indexoid );
+ 	if ( !GinGetUseFastUpdate(irel) )
+ 	{
+ 		/*
+ 		 * Index isn't in fast update mode 
+ 		 */
+ 
+ 		RelationClose(irel);
+ 		PG_RETURN_VOID();
+ 	}
+ 	RelationClose(irel);
+ 
+ 	/*
+ 	 * Get table's oid
+ 	 */
+ 	rte = planner_rt_fetch(index->rel->relid, root);
+ 	Assert(rte->rtekind == RTE_RELATION);
+ 	Assert(rte->relid != InvalidOid);
+ 
+ 	/*
+ 	 * Add cost of scaning pending list and we would like to 
+ 	 * forbid index scan if n_pending_tuples is too high and 
+ 	 * there is a risk to make tidbitmap GinScanOpaque->scanFastTuples lossy.
+ 	 */
+ 
+ 	n_pending_tuples = DatumGetInt64(DirectFunctionCall1(
+ 							pg_stat_get_fresh_inserted_tuples,
+ 							ObjectIdGetDatum(rte->relid)));
+ 
+ 	if ( n_pending_tuples <= 0 )
+ 	{
+ 		/*
+ 		 * There is no pending pages
+ 		 */
+ 		pendingPageCost = 0.0;
+ 	}
+ 	else if ( *indexSelectivity * n_pending_tuples >= tbm_max_non_lossy( work_mem * 1024L ) )
+ 	{
+ 		/*
+ 		 * Number of tuples found in pending list is too big
+ 		 * to keep tidbitmap non-lossy, so forbid index scan.
+ 		 * tbm_max_non_lossy returns theoretical maximum 
+ 		 * of tuple's number with guaranteed non-lossy tidbitmap
+ 		 */
+ 		pendingPageCost = disable_cost;
+ 	}
+ 	else
+ 	{
+ 		double  pending_pages_fetched;
+ 
+ 		if ( index->tuples <= (double)n_pending_tuples )
+ 		{
+ 			/*
+ 			 * Suppose, all indexed values placed in pending pages
+ 			 */
+ 			pending_pages_fetched = index->pages;
+ 		}
+ 		else
+ 		{
+ 			/*
+ 			 * Number of pending pages is unknown, so we estimate it by
+ 			 * fraction of number of pending tuples from total number of
+ 			 * tuples
+ 			 */
+ 			pending_pages_fetched = index->pages * ( (double)n_pending_tuples / index->tuples );
+ 		}
+ 
+ 		pendingPageCost = pending_pages_fetched * seq_page_cost; 
+ 	}
+ 
+ /*
+ elog(NOTICE,"pendingPageCost:%f n_pending_tuples:%lld (limit:%ld) index->pages:%u index->tuples:%f", 
+ 					pendingPageCost, 
+ 					n_pending_tuples, 
+ 					tbm_max_non_lossy( work_mem * 1024L ),
+ 					index->pages,
+ 					index->tuples
+ 					);
+ */
+ 	*indexStartupCost += pendingPageCost;
+ 	*indexTotalCost   += pendingPageCost;
+ 	
  	PG_RETURN_VOID();
  }
diff -c -r -N src.orig/include/access/gin.h src/include/access/gin.h
*** src.orig/include/access/gin.h	2009-02-04 18:13:36.000000000 +0300
--- src/include/access/gin.h	2009-02-04 18:43:51.000000000 +0300
***************
*** 21,26 ****
--- 21,27 ----
  #include "storage/buf.h"
  #include "storage/off.h"
  #include "storage/relfilenode.h"
+ #include "utils/rel.h"
  
  
  /*
***************
*** 46,62 ****
  	OffsetNumber maxoff;		/* number entries on GIN_DATA page: number of
  								 * heap ItemPointer on GIN_DATA|GIN_LEAF page
  								 * and number of records on GIN_DATA &
! 								 * ~GIN_LEAF page */
  	uint16		flags;			/* see bit definitions below */
  } GinPageOpaqueData;
  
  typedef GinPageOpaqueData *GinPageOpaque;
  
! #define GIN_ROOT_BLKNO	(0)
  
  #define GIN_DATA		  (1 << 0)
  #define GIN_LEAF		  (1 << 1)
  #define GIN_DELETED		  (1 << 2)
  
  /*
   * Works on page
--- 47,93 ----
  	OffsetNumber maxoff;		/* number entries on GIN_DATA page: number of
  								 * heap ItemPointer on GIN_DATA|GIN_LEAF page
  								 * and number of records on GIN_DATA &
! 								 * ~GIN_LEAF page. On GIN_LIST number of heap tuples.*/
  	uint16		flags;			/* see bit definitions below */
  } GinPageOpaqueData;
  
  typedef GinPageOpaqueData *GinPageOpaque;
  
! #define GIN_METAPAGE_BLKNO	(0)
! #define GIN_ROOT_BLKNO	(1)
  
  #define GIN_DATA		  (1 << 0)
  #define GIN_LEAF		  (1 << 1)
  #define GIN_DELETED		  (1 << 2)
+ #define GIN_META		  (1 << 3)
+ #define GIN_LIST		  (1 << 4)
+ #define GIN_LIST_FULLROW  (1 << 5)   /* makes sense only on GIN_LIST page */
+ 
+ typedef struct GinMetaPageData
+ {
+ 	/*
+ 	 * Pointers to head and tail of list of GIN_LIST pages.  These store
+ 	 * fast-inserted entries that haven't yet been moved into the regular
+ 	 * GIN structure.
+ 	 */
+ 	BlockNumber			head;
+ 	BlockNumber         tail;
+ 
+ 	/*
+ 	 * Free space in bytes in the list's tail page.
+ 	 */
+ 	uint32				tailFreeSize;
+ 
+ 	/*
+ 	 * Store both number of pages and number of heap tuples
+ 	 * in pending list.
+ 	 */
+ 	BlockNumber			nPendingPages;
+ 	int64				nPendingHeapTuples;
+ } GinMetaPageData;
+ 
+ #define GinPageGetMeta(p) \
+ 	((GinMetaPageData *) PageGetContents(p))
  
  /*
   * Works on page
***************
*** 68,73 ****
--- 99,106 ----
  #define GinPageSetNonLeaf(page)    ( GinPageGetOpaque(page)->flags &= ~GIN_LEAF )
  #define GinPageIsData(page)    ( GinPageGetOpaque(page)->flags & GIN_DATA )
  #define GinPageSetData(page)   ( GinPageGetOpaque(page)->flags |= GIN_DATA )
+ #define GinPageHasFullRow(page)    ( GinPageGetOpaque(page)->flags & GIN_LIST_FULLROW )
+ #define GinPageSetFullRow(page)   ( GinPageGetOpaque(page)->flags |= GIN_LIST_FULLROW )
  
  #define GinPageIsDeleted(page) ( GinPageGetOpaque(page)->flags & GIN_DELETED)
  #define GinPageSetDeleted(page)    ( GinPageGetOpaque(page)->flags |= GIN_DELETED)
***************
*** 135,140 ****
--- 168,187 ----
  	 - GinPageGetOpaque(page)->maxoff * GinSizeOfItem(page) \
  	 - MAXALIGN(sizeof(GinPageOpaqueData)))
  
+ /*
+  * storage type for GIN's options.  Must be upward compatible with
+  * StdRdOptions, since we might call RelationGetFillFactor().
+  */
+ typedef struct GinOptions
+ {
+ 	int32       vl_len_;        /* varlena header (do not touch directly!) */
+ 	bool		useFastUpdate;	/* use fast updates? */
+ } GinOptions;
+ 
+ #define GIN_DEFAULT_USE_FASTUPDATE  true
+ #define GinGetUseFastUpdate(relation) \
+     ((relation)->rd_options ? \
+ 	 ((GinOptions *) (relation)->rd_options)->useFastUpdate : GIN_DEFAULT_USE_FASTUPDATE)
  
  #define GIN_UNLOCK	BUFFER_LOCK_UNLOCK
  #define GIN_SHARE	BUFFER_LOCK_SHARE
***************
*** 234,245 ****
--- 281,329 ----
  	BlockNumber rightLink;
  } ginxlogDeletePage;
  
+ 
+ #define XLOG_GIN_UPDATE_META_PAGE 0x60
+ 
+ typedef struct ginxlogUpdateMeta
+ {
+ 	RelFileNode     node;
+ 	GinMetaPageData metadata;
+ 	BlockNumber     prevTail;
+ 	BlockNumber     newRightlink;
+ 	int32           ntuples; /* if ntuples > 0 then metadata.tail was updated with
+ 								that tuples else new sub list was inserted */
+ 	/* follows array of inserted tuples */
+ } ginxlogUpdateMeta;
+ 
+ #define XLOG_GIN_INSERT_LISTPAGE  0x70
+ 
+ typedef struct ginxlogInsertListPage
+ {
+ 	RelFileNode     node;
+ 	BlockNumber     blkno;
+ 	BlockNumber     rightlink;
+ 	int32           ntuples;
+ 	/* follows array of inserted tuples */
+ } ginxlogInsertListPage;
+ 
+ #define XLOG_GIN_DELETE_LISTPAGE  0x80
+ 
+ #define NDELETE_AT_ONCE (16)
+ typedef struct ginxlogDeleteListPages
+ {
+ 	RelFileNode     node;
+ 	GinMetaPageData metadata;
+ 	int32           ndeleted;
+ 	BlockNumber     toDelete[ NDELETE_AT_ONCE ];
+ } ginxlogDeleteListPages;
+ 
  /* ginutil.c */
  extern Datum ginoptions(PG_FUNCTION_ARGS);
  extern void initGinState(GinState *state, Relation index);
  extern Buffer GinNewBuffer(Relation index);
  extern void GinInitBuffer(Buffer b, uint32 f);
  extern void GinInitPage(Page page, uint32 f, Size pageSize);
+ extern void GinInitMetabuffer(Buffer b);
  extern int	compareEntries(GinState *ginstate, OffsetNumber attnum, Datum a, Datum b);
  extern int	compareAttEntries(GinState *ginstate, OffsetNumber attnum_a, Datum a, 
  												  OffsetNumber attnum_b, Datum b);
***************
*** 249,257 ****
--- 333,344 ----
  
  extern Datum gin_index_getattr(GinState *ginstate, IndexTuple tuple);
  extern OffsetNumber gintuple_get_attrnum(GinState *ginstate, IndexTuple tuple);
+ 
  /* gininsert.c */
  extern Datum ginbuild(PG_FUNCTION_ARGS);
  extern Datum gininsert(PG_FUNCTION_ARGS);
+ extern void ginEntryInsert(Relation index, GinState *ginstate, OffsetNumber attnum, Datum value,
+ 							ItemPointerData *items, uint32 nitem, bool isBuild);
  
  /* ginxlog.c */
  extern void gin_redo(XLogRecPtr lsn, XLogRecord *record);
***************
*** 426,431 ****
--- 513,523 ----
  	uint32		nkeys;
  	bool		isVoidRes;		/* true if ginstate.extractQueryFn guarantees
  								 * that nothing will be found */
+ 	bool				isScanFastInsert; /* scan process in scanning fast update pages */
+ 	TIDBitmap			*scanFastTuples;
+ 	TBMIterator			*scanFastIterator;
+ 	TBMIterateResult	*scanFastResult;
+ 	OffsetNumber		scanFastOffset;
  } GinScanOpaqueData;
  
  typedef GinScanOpaqueData *GinScanOpaque;
***************
*** 489,492 ****
--- 581,603 ----
  				  OffsetNumber attnum, Datum *entries, int32 nentry);
  extern ItemPointerData *ginGetEntry(BuildAccumulator *accum, OffsetNumber *attnum, Datum *entry, uint32 *n);
  
+ /* ginfast.c */
+ 
+ typedef struct GinTupleCollector {
+ 	IndexTuple	*tuples;
+ 	uint32		 ntuples;
+ 	uint32		 lentuples;
+ 	uint32		 sumsize;
+ } GinTupleCollector; 
+ 
+ extern void ginHeapTupleFastInsert(Relation index, GinState *ginstate, GinTupleCollector *collector);  
+ extern uint32 ginHeapTupleFastCollect(Relation index, GinState *ginstate,
+ 					GinTupleCollector *collector,
+ 					OffsetNumber attnum, Datum value, ItemPointer item);
+ 
+ #define GinListPageSize   \
+     ( BLCKSZ - SizeOfPageHeaderData - MAXALIGN(sizeof(GinPageOpaqueData)) )
+ 
+ extern void ginInsertCleanup(Relation index, GinState *ginstate, IndexBulkDeleteResult *stats);
+ 
  #endif
diff -c -r -N src.orig/include/catalog/pg_proc.h src/include/catalog/pg_proc.h
*** src.orig/include/catalog/pg_proc.h	2009-02-04 18:13:35.000000000 +0300
--- src/include/catalog/pg_proc.h	2009-02-04 18:23:22.000000000 +0300
***************
*** 2928,2933 ****
--- 2928,2935 ----
  DESCR("statistics: number of tuples deleted");
  DATA(insert OID = 1972 (  pg_stat_get_tuples_hot_updated PGNSP PGUID 12 1 0 0 f f f t f s 1 0 20 "26" _null_ _null_ _null_ _null_ pg_stat_get_tuples_hot_updated _null_ _null_ _null_ ));
  DESCR("statistics: number of tuples hot updated");
+ DATA(insert OID = 2316 (  pg_stat_get_fresh_inserted_tuples	PGNSP PGUID 12 1 0 0 f f f t f s 1 0 20 "26" _null_ _null_ _null_ _null_ pg_stat_get_fresh_inserted_tuples _null_ _null_ _null_ ));
+ DESCR("statistics: number of inserted tuples since last vacuum");
  DATA(insert OID = 2878 (  pg_stat_get_live_tuples	PGNSP PGUID 12 1 0 0 f f f t f s 1 0 20 "26" _null_ _null_ _null_ _null_ pg_stat_get_live_tuples _null_ _null_ _null_ ));
  DESCR("statistics: number of live tuples");
  DATA(insert OID = 2879 (  pg_stat_get_dead_tuples	PGNSP PGUID 12 1 0 0 f f f t f s 1 0 20 "26" _null_ _null_ _null_ _null_ pg_stat_get_dead_tuples _null_ _null_ _null_ ));
diff -c -r -N src.orig/include/nodes/tidbitmap.h src/include/nodes/tidbitmap.h
*** src.orig/include/nodes/tidbitmap.h	2009-02-04 18:13:35.000000000 +0300
--- src/include/nodes/tidbitmap.h	2009-02-04 18:23:22.000000000 +0300
***************
*** 52,57 ****
--- 52,60 ----
  extern void tbm_add_tuples(TIDBitmap *tbm,
  						   const ItemPointer tids, int ntids,
  						   bool recheck);
+ extern bool tbm_check_tuple(TIDBitmap *tbm, const ItemPointer tid);
+ extern bool	tbm_has_lossy(TIDBitmap *tbm);
+ extern long tbm_max_non_lossy(long maxbytes);
  
  extern void tbm_union(TIDBitmap *a, const TIDBitmap *b);
  extern void tbm_intersect(TIDBitmap *a, const TIDBitmap *b);
diff -c -r -N src.orig/include/pgstat.h src/include/pgstat.h
*** src.orig/include/pgstat.h	2009-02-04 18:13:37.000000000 +0300
--- src/include/pgstat.h	2009-02-04 18:45:38.000000000 +0300
***************
*** 476,481 ****
--- 476,483 ----
  	PgStat_Counter tuples_deleted;
  	PgStat_Counter tuples_hot_updated;
  
+ 	PgStat_Counter n_inserted_tuples; /* number of non-hot inserted tuples
+ 									   * since last vacuum */ 
  	PgStat_Counter n_live_tuples;
  	PgStat_Counter n_dead_tuples;
  	PgStat_Counter last_anl_tuples;
***************
*** 721,724 ****
--- 723,729 ----
  extern int	pgstat_fetch_stat_numbackends(void);
  extern PgStat_GlobalStats *pgstat_fetch_global(void);
  
+ 
+ extern Datum pg_stat_get_fresh_inserted_tuples(PG_FUNCTION_ARGS);
+ 
  #endif   /* PGSTAT_H */
diff -c -r -N src.orig/test/regress/expected/rules.out src/test/regress/expected/rules.out
*** src.orig/test/regress/expected/rules.out	2009-02-04 18:13:38.000000000 +0300
--- src/test/regress/expected/rules.out	2009-02-04 18:23:23.000000000 +0300
***************
*** 1291,1304 ****
   pg_shadow                | SELECT pg_authid.rolname AS usename, pg_authid.oid AS usesysid, pg_authid.rolcreatedb AS usecreatedb, pg_authid.rolsuper AS usesuper, pg_authid.rolcatupdate AS usecatupd, pg_authid.rolpassword AS passwd, (pg_authid.rolvaliduntil)::abstime AS valuntil, pg_authid.rolconfig AS useconfig FROM pg_authid WHERE pg_authid.rolcanlogin;
   pg_stat_activity         | SELECT s.datid, d.datname, s.procpid, s.usesysid, u.rolname AS usename, s.current_query, s.waiting, s.xact_start, s.query_start, s.backend_start, s.client_addr, s.client_port FROM pg_database d, pg_stat_get_activity(NULL::integer) s(datid, procpid, usesysid, current_query, waiting, xact_start, query_start, backend_start, client_addr, client_port), pg_authid u WHERE ((s.datid = d.oid) AND (s.usesysid = u.oid));
   pg_stat_all_indexes      | SELECT c.oid AS relid, i.oid AS indexrelid, n.nspname AS schemaname, c.relname, i.relname AS indexrelname, pg_stat_get_numscans(i.oid) AS idx_scan, pg_stat_get_tuples_returned(i.oid) AS idx_tup_read, pg_stat_get_tuples_fetched(i.oid) AS idx_tup_fetch FROM (((pg_class c JOIN pg_index x ON ((c.oid = x.indrelid))) JOIN pg_class i ON ((i.oid = x.indexrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"]));
!  pg_stat_all_tables       | SELECT c.oid AS relid, n.nspname AS schemaname, c.relname, pg_stat_get_numscans(c.oid) AS seq_scan, pg_stat_get_tuples_returned(c.oid) AS seq_tup_read, (sum(pg_stat_get_numscans(i.indexrelid)))::bigint AS idx_scan, ((sum(pg_stat_get_tuples_fetched(i.indexrelid)))::bigint + pg_stat_get_tuples_fetched(c.oid)) AS idx_tup_fetch, pg_stat_get_tuples_inserted(c.oid) AS n_tup_ins, pg_stat_get_tuples_updated(c.oid) AS n_tup_upd, pg_stat_get_tuples_deleted(c.oid) AS n_tup_del, pg_stat_get_tuples_hot_updated(c.oid) AS n_tup_hot_upd, pg_stat_get_live_tuples(c.oid) AS n_live_tup, pg_stat_get_dead_tuples(c.oid) AS n_dead_tup, pg_stat_get_last_vacuum_time(c.oid) AS last_vacuum, pg_stat_get_last_autovacuum_time(c.oid) AS last_autovacuum, pg_stat_get_last_analyze_time(c.oid) AS last_analyze, pg_stat_get_last_autoanalyze_time(c.oid) AS last_autoanalyze FROM ((pg_class c LEFT JOIN pg_index i ON ((c.oid = i.indrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"])) GROUP BY c.oid, n.nspname, c.relname;
   pg_stat_bgwriter         | SELECT pg_stat_get_bgwriter_timed_checkpoints() AS checkpoints_timed, pg_stat_get_bgwriter_requested_checkpoints() AS checkpoints_req, pg_stat_get_bgwriter_buf_written_checkpoints() AS buffers_checkpoint, pg_stat_get_bgwriter_buf_written_clean() AS buffers_clean, pg_stat_get_bgwriter_maxwritten_clean() AS maxwritten_clean, pg_stat_get_buf_written_backend() AS buffers_backend, pg_stat_get_buf_alloc() AS buffers_alloc;
   pg_stat_database         | SELECT d.oid AS datid, d.datname, pg_stat_get_db_numbackends(d.oid) AS numbackends, pg_stat_get_db_xact_commit(d.oid) AS xact_commit, pg_stat_get_db_xact_rollback(d.oid) AS xact_rollback, (pg_stat_get_db_blocks_fetched(d.oid) - pg_stat_get_db_blocks_hit(d.oid)) AS blks_read, pg_stat_get_db_blocks_hit(d.oid) AS blks_hit, pg_stat_get_db_tuples_returned(d.oid) AS tup_returned, pg_stat_get_db_tuples_fetched(d.oid) AS tup_fetched, pg_stat_get_db_tuples_inserted(d.oid) AS tup_inserted, pg_stat_get_db_tuples_updated(d.oid) AS tup_updated, pg_stat_get_db_tuples_deleted(d.oid) AS tup_deleted FROM pg_database d;
   pg_stat_sys_indexes      | SELECT pg_stat_all_indexes.relid, pg_stat_all_indexes.indexrelid, pg_stat_all_indexes.schemaname, pg_stat_all_indexes.relname, pg_stat_all_indexes.indexrelname, pg_stat_all_indexes.idx_scan, pg_stat_all_indexes.idx_tup_read, pg_stat_all_indexes.idx_tup_fetch FROM pg_stat_all_indexes WHERE ((pg_stat_all_indexes.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_indexes.schemaname ~ '^pg_toast'::text));
!  pg_stat_sys_tables       | SELECT pg_stat_all_tables.relid, pg_stat_all_tables.schemaname, pg_stat_all_tables.relname, pg_stat_all_tables.seq_scan, pg_stat_all_tables.seq_tup_read, pg_stat_all_tables.idx_scan, pg_stat_all_tables.idx_tup_fetch, pg_stat_all_tables.n_tup_ins, pg_stat_all_tables.n_tup_upd, pg_stat_all_tables.n_tup_del, pg_stat_all_tables.n_tup_hot_upd, pg_stat_all_tables.n_live_tup, pg_stat_all_tables.n_dead_tup, pg_stat_all_tables.last_vacuum, pg_stat_all_tables.last_autovacuum, pg_stat_all_tables.last_analyze, pg_stat_all_tables.last_autoanalyze FROM pg_stat_all_tables WHERE ((pg_stat_all_tables.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_tables.schemaname ~ '^pg_toast'::text));
   pg_stat_user_functions   | SELECT p.oid AS funcid, n.nspname AS schemaname, p.proname AS funcname, pg_stat_get_function_calls(p.oid) AS calls, (pg_stat_get_function_time(p.oid) / 1000) AS total_time, (pg_stat_get_function_self_time(p.oid) / 1000) AS self_time FROM (pg_proc p LEFT JOIN pg_namespace n ON ((n.oid = p.pronamespace))) WHERE ((p.prolang <> (12)::oid) AND (pg_stat_get_function_calls(p.oid) IS NOT NULL));
   pg_stat_user_indexes     | SELECT pg_stat_all_indexes.relid, pg_stat_all_indexes.indexrelid, pg_stat_all_indexes.schemaname, pg_stat_all_indexes.relname, pg_stat_all_indexes.indexrelname, pg_stat_all_indexes.idx_scan, pg_stat_all_indexes.idx_tup_read, pg_stat_all_indexes.idx_tup_fetch FROM pg_stat_all_indexes WHERE ((pg_stat_all_indexes.schemaname <> ALL (ARRAY['pg_catalog'::name, 'information_schema'::name])) AND (pg_stat_all_indexes.schemaname !~ '^pg_toast'::text));
!  pg_stat_user_tables      | SELECT pg_stat_all_tables.relid, pg_stat_all_tables.schemaname, pg_stat_all_tables.relname, pg_stat_all_tables.seq_scan, pg_stat_all_tables.seq_tup_read, pg_stat_all_tables.idx_scan, pg_stat_all_tables.idx_tup_fetch, pg_stat_all_tables.n_tup_ins, pg_stat_all_tables.n_tup_upd, pg_stat_all_tables.n_tup_del, pg_stat_all_tables.n_tup_hot_upd, pg_stat_all_tables.n_live_tup, pg_stat_all_tables.n_dead_tup, pg_stat_all_tables.last_vacuum, pg_stat_all_tables.last_autovacuum, pg_stat_all_tables.last_analyze, pg_stat_all_tables.last_autoanalyze FROM pg_stat_all_tables WHERE ((pg_stat_all_tables.schemaname <> ALL (ARRAY['pg_catalog'::name, 'information_schema'::name])) AND (pg_stat_all_tables.schemaname !~ '^pg_toast'::text));
   pg_statio_all_indexes    | SELECT c.oid AS relid, i.oid AS indexrelid, n.nspname AS schemaname, c.relname, i.relname AS indexrelname, (pg_stat_get_blocks_fetched(i.oid) - pg_stat_get_blocks_hit(i.oid)) AS idx_blks_read, pg_stat_get_blocks_hit(i.oid) AS idx_blks_hit FROM (((pg_class c JOIN pg_index x ON ((c.oid = x.indrelid))) JOIN pg_class i ON ((i.oid = x.indexrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"]));
   pg_statio_all_sequences  | SELECT c.oid AS relid, n.nspname AS schemaname, c.relname, (pg_stat_get_blocks_fetched(c.oid) - pg_stat_get_blocks_hit(c.oid)) AS blks_read, pg_stat_get_blocks_hit(c.oid) AS blks_hit FROM (pg_class c LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = 'S'::"char");
   pg_statio_all_tables     | SELECT c.oid AS relid, n.nspname AS schemaname, c.relname, (pg_stat_get_blocks_fetched(c.oid) - pg_stat_get_blocks_hit(c.oid)) AS heap_blks_read, pg_stat_get_blocks_hit(c.oid) AS heap_blks_hit, (sum((pg_stat_get_blocks_fetched(i.indexrelid) - pg_stat_get_blocks_hit(i.indexrelid))))::bigint AS idx_blks_read, (sum(pg_stat_get_blocks_hit(i.indexrelid)))::bigint AS idx_blks_hit, (pg_stat_get_blocks_fetched(t.oid) - pg_stat_get_blocks_hit(t.oid)) AS toast_blks_read, pg_stat_get_blocks_hit(t.oid) AS toast_blks_hit, (pg_stat_get_blocks_fetched(x.oid) - pg_stat_get_blocks_hit(x.oid)) AS tidx_blks_read, pg_stat_get_blocks_hit(x.oid) AS tidx_blks_hit FROM ((((pg_class c LEFT JOIN pg_index i ON ((c.oid = i.indrelid))) LEFT JOIN pg_class t ON ((c.reltoastrelid = t.oid))) LEFT JOIN pg_class x ON ((t.reltoastidxid = x.oid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"])) GROUP BY c.oid, n.nspname, c.relname, t.oid, x.oid;
--- 1291,1304 ----
   pg_shadow                | SELECT pg_authid.rolname AS usename, pg_authid.oid AS usesysid, pg_authid.rolcreatedb AS usecreatedb, pg_authid.rolsuper AS usesuper, pg_authid.rolcatupdate AS usecatupd, pg_authid.rolpassword AS passwd, (pg_authid.rolvaliduntil)::abstime AS valuntil, pg_authid.rolconfig AS useconfig FROM pg_authid WHERE pg_authid.rolcanlogin;
   pg_stat_activity         | SELECT s.datid, d.datname, s.procpid, s.usesysid, u.rolname AS usename, s.current_query, s.waiting, s.xact_start, s.query_start, s.backend_start, s.client_addr, s.client_port FROM pg_database d, pg_stat_get_activity(NULL::integer) s(datid, procpid, usesysid, current_query, waiting, xact_start, query_start, backend_start, client_addr, client_port), pg_authid u WHERE ((s.datid = d.oid) AND (s.usesysid = u.oid));
   pg_stat_all_indexes      | SELECT c.oid AS relid, i.oid AS indexrelid, n.nspname AS schemaname, c.relname, i.relname AS indexrelname, pg_stat_get_numscans(i.oid) AS idx_scan, pg_stat_get_tuples_returned(i.oid) AS idx_tup_read, pg_stat_get_tuples_fetched(i.oid) AS idx_tup_fetch FROM (((pg_class c JOIN pg_index x ON ((c.oid = x.indrelid))) JOIN pg_class i ON ((i.oid = x.indexrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"]));
!  pg_stat_all_tables       | SELECT c.oid AS relid, n.nspname AS schemaname, c.relname, pg_stat_get_numscans(c.oid) AS seq_scan, pg_stat_get_tuples_returned(c.oid) AS seq_tup_read, (sum(pg_stat_get_numscans(i.indexrelid)))::bigint AS idx_scan, ((sum(pg_stat_get_tuples_fetched(i.indexrelid)))::bigint + pg_stat_get_tuples_fetched(c.oid)) AS idx_tup_fetch, pg_stat_get_tuples_inserted(c.oid) AS n_tup_ins, pg_stat_get_tuples_updated(c.oid) AS n_tup_upd, pg_stat_get_tuples_deleted(c.oid) AS n_tup_del, pg_stat_get_tuples_hot_updated(c.oid) AS n_tup_hot_upd, pg_stat_get_fresh_inserted_tuples(c.oid) AS n_fresh_tup, pg_stat_get_live_tuples(c.oid) AS n_live_tup, pg_stat_get_dead_tuples(c.oid) AS n_dead_tup, pg_stat_get_last_vacuum_time(c.oid) AS last_vacuum, pg_stat_get_last_autovacuum_time(c.oid) AS last_autovacuum, pg_stat_get_last_analyze_time(c.oid) AS last_analyze, pg_stat_get_last_autoanalyze_time(c.oid) AS last_autoanalyze FROM ((pg_class c LEFT JOIN pg_index i ON ((c.oid = i.indrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"])) GROUP BY c.oid, n.nspname, c.relname;
   pg_stat_bgwriter         | SELECT pg_stat_get_bgwriter_timed_checkpoints() AS checkpoints_timed, pg_stat_get_bgwriter_requested_checkpoints() AS checkpoints_req, pg_stat_get_bgwriter_buf_written_checkpoints() AS buffers_checkpoint, pg_stat_get_bgwriter_buf_written_clean() AS buffers_clean, pg_stat_get_bgwriter_maxwritten_clean() AS maxwritten_clean, pg_stat_get_buf_written_backend() AS buffers_backend, pg_stat_get_buf_alloc() AS buffers_alloc;
   pg_stat_database         | SELECT d.oid AS datid, d.datname, pg_stat_get_db_numbackends(d.oid) AS numbackends, pg_stat_get_db_xact_commit(d.oid) AS xact_commit, pg_stat_get_db_xact_rollback(d.oid) AS xact_rollback, (pg_stat_get_db_blocks_fetched(d.oid) - pg_stat_get_db_blocks_hit(d.oid)) AS blks_read, pg_stat_get_db_blocks_hit(d.oid) AS blks_hit, pg_stat_get_db_tuples_returned(d.oid) AS tup_returned, pg_stat_get_db_tuples_fetched(d.oid) AS tup_fetched, pg_stat_get_db_tuples_inserted(d.oid) AS tup_inserted, pg_stat_get_db_tuples_updated(d.oid) AS tup_updated, pg_stat_get_db_tuples_deleted(d.oid) AS tup_deleted FROM pg_database d;
   pg_stat_sys_indexes      | SELECT pg_stat_all_indexes.relid, pg_stat_all_indexes.indexrelid, pg_stat_all_indexes.schemaname, pg_stat_all_indexes.relname, pg_stat_all_indexes.indexrelname, pg_stat_all_indexes.idx_scan, pg_stat_all_indexes.idx_tup_read, pg_stat_all_indexes.idx_tup_fetch FROM pg_stat_all_indexes WHERE ((pg_stat_all_indexes.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_indexes.schemaname ~ '^pg_toast'::text));
!  pg_stat_sys_tables       | SELECT pg_stat_all_tables.relid, pg_stat_all_tables.schemaname, pg_stat_all_tables.relname, pg_stat_all_tables.seq_scan, pg_stat_all_tables.seq_tup_read, pg_stat_all_tables.idx_scan, pg_stat_all_tables.idx_tup_fetch, pg_stat_all_tables.n_tup_ins, pg_stat_all_tables.n_tup_upd, pg_stat_all_tables.n_tup_del, pg_stat_all_tables.n_tup_hot_upd, pg_stat_all_tables.n_fresh_tup, pg_stat_all_tables.n_live_tup, pg_stat_all_tables.n_dead_tup, pg_stat_all_tables.last_vacuum, pg_stat_all_tables.last_autovacuum, pg_stat_all_tables.last_analyze, pg_stat_all_tables.last_autoanalyze FROM pg_stat_all_tables WHERE ((pg_stat_all_tables.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_tables.schemaname ~ '^pg_toast'::text));
   pg_stat_user_functions   | SELECT p.oid AS funcid, n.nspname AS schemaname, p.proname AS funcname, pg_stat_get_function_calls(p.oid) AS calls, (pg_stat_get_function_time(p.oid) / 1000) AS total_time, (pg_stat_get_function_self_time(p.oid) / 1000) AS self_time FROM (pg_proc p LEFT JOIN pg_namespace n ON ((n.oid = p.pronamespace))) WHERE ((p.prolang <> (12)::oid) AND (pg_stat_get_function_calls(p.oid) IS NOT NULL));
   pg_stat_user_indexes     | SELECT pg_stat_all_indexes.relid, pg_stat_all_indexes.indexrelid, pg_stat_all_indexes.schemaname, pg_stat_all_indexes.relname, pg_stat_all_indexes.indexrelname, pg_stat_all_indexes.idx_scan, pg_stat_all_indexes.idx_tup_read, pg_stat_all_indexes.idx_tup_fetch FROM pg_stat_all_indexes WHERE ((pg_stat_all_indexes.schemaname <> ALL (ARRAY['pg_catalog'::name, 'information_schema'::name])) AND (pg_stat_all_indexes.schemaname !~ '^pg_toast'::text));
!  pg_stat_user_tables      | SELECT pg_stat_all_tables.relid, pg_stat_all_tables.schemaname, pg_stat_all_tables.relname, pg_stat_all_tables.seq_scan, pg_stat_all_tables.seq_tup_read, pg_stat_all_tables.idx_scan, pg_stat_all_tables.idx_tup_fetch, pg_stat_all_tables.n_tup_ins, pg_stat_all_tables.n_tup_upd, pg_stat_all_tables.n_tup_del, pg_stat_all_tables.n_tup_hot_upd, pg_stat_all_tables.n_fresh_tup, pg_stat_all_tables.n_live_tup, pg_stat_all_tables.n_dead_tup, pg_stat_all_tables.last_vacuum, pg_stat_all_tables.last_autovacuum, pg_stat_all_tables.last_analyze, pg_stat_all_tables.last_autoanalyze FROM pg_stat_all_tables WHERE ((pg_stat_all_tables.schemaname <> ALL (ARRAY['pg_catalog'::name, 'information_schema'::name])) AND (pg_stat_all_tables.schemaname !~ '^pg_toast'::text));
   pg_statio_all_indexes    | SELECT c.oid AS relid, i.oid AS indexrelid, n.nspname AS schemaname, c.relname, i.relname AS indexrelname, (pg_stat_get_blocks_fetched(i.oid) - pg_stat_get_blocks_hit(i.oid)) AS idx_blks_read, pg_stat_get_blocks_hit(i.oid) AS idx_blks_hit FROM (((pg_class c JOIN pg_index x ON ((c.oid = x.indrelid))) JOIN pg_class i ON ((i.oid = x.indexrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"]));
   pg_statio_all_sequences  | SELECT c.oid AS relid, n.nspname AS schemaname, c.relname, (pg_stat_get_blocks_fetched(c.oid) - pg_stat_get_blocks_hit(c.oid)) AS blks_read, pg_stat_get_blocks_hit(c.oid) AS blks_hit FROM (pg_class c LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = 'S'::"char");
   pg_statio_all_tables     | SELECT c.oid AS relid, n.nspname AS schemaname, c.relname, (pg_stat_get_blocks_fetched(c.oid) - pg_stat_get_blocks_hit(c.oid)) AS heap_blks_read, pg_stat_get_blocks_hit(c.oid) AS heap_blks_hit, (sum((pg_stat_get_blocks_fetched(i.indexrelid) - pg_stat_get_blocks_hit(i.indexrelid))))::bigint AS idx_blks_read, (sum(pg_stat_get_blocks_hit(i.indexrelid)))::bigint AS idx_blks_hit, (pg_stat_get_blocks_fetched(t.oid) - pg_stat_get_blocks_hit(t.oid)) AS toast_blks_read, pg_stat_get_blocks_hit(t.oid) AS toast_blks_hit, (pg_stat_get_blocks_fetched(x.oid) - pg_stat_get_blocks_hit(x.oid)) AS tidx_blks_read, pg_stat_get_blocks_hit(x.oid) AS tidx_blks_hit FROM ((((pg_class c LEFT JOIN pg_index i ON ((c.oid = i.indrelid))) LEFT JOIN pg_class t ON ((c.reltoastrelid = t.oid))) LEFT JOIN pg_class x ON ((t.reltoastidxid = x.oid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"])) GROUP BY c.oid, n.nspname, c.relname, t.oid, x.oid;
diff -c -r -N src.orig/test/regression.out src/test/regression.out
*** src.orig/test/regression.out	1970-01-01 03:00:00.000000000 +0300
--- src/test/regression.out	2009-02-04 18:23:23.000000000 +0300
***************
*** 0 ****
--- 1,2 ----
+ parallel group (17 tests):  int2 char boolean varchar name money enum uuid txid numeric bit float8 float4 oid int8 int4 text
+      boolean              ... 
\ No newline at end of file
diff -c -r -N doc.orig/src/sgml/catalogs.sgml doc/src/sgml/catalogs.sgml
*** doc.orig/src/sgml/catalogs.sgml	2009-02-04 18:13:39.000000000 +0300
--- doc/src/sgml/catalogs.sgml	2009-02-04 18:23:23.000000000 +0300
***************
*** 551,556 ****
--- 551,563 ----
       </row>
  
       <row>
+       <entry><structfield>aminsertcleanup</structfield></entry>
+       <entry><type>regproc</type></entry>
+       <entry><literal><link linkend="catalog-pg-proc"><structname>pg_proc</structname></link>.oid</literal></entry>
+       <entry>Post-<command>INSERT</command> cleanup finction (optional)</entry>
+      </row>
+ 
+      <row>
        <entry><structfield>amvacuumcleanup</structfield></entry>
        <entry><type>regproc</type></entry>
        <entry><literal><link linkend="catalog-pg-proc"><structname>pg_proc</structname></link>.oid</literal></entry>
diff -c -r -N doc.orig/src/sgml/config.sgml doc/src/sgml/config.sgml
*** doc.orig/src/sgml/config.sgml	2009-02-04 18:13:39.000000000 +0300
--- doc/src/sgml/config.sgml	2009-02-04 18:23:23.000000000 +0300
***************
*** 3550,3555 ****
--- 3550,3560 ----
          This setting can be overridden for individual tables by entries in
          <structname>pg_autovacuum</>.
         </para>
+        <para>
+         This parameter affects on vacuuming a table with <acronym>GIN</acronym>
+         index, it also specifies the minimum number of inserted or updated 
+         tuples needed to trigger a <command>VACUUM</>  on thos table.
+        </para>
        </listitem>
       </varlistentry>
  
diff -c -r -N doc.orig/src/sgml/gin.sgml doc/src/sgml/gin.sgml
*** doc.orig/src/sgml/gin.sgml	2009-02-04 18:13:39.000000000 +0300
--- doc/src/sgml/gin.sgml	2009-02-04 18:23:23.000000000 +0300
***************
*** 188,196 ****
    list of heap pointers (PL, posting list) if the list is small enough.
   </para>
  
   <sect2 id="gin-partial-match">
    <title>Partial match algorithm</title>
!   
    <para>
     GIN can support <quote>partial match</> queries, in which the query
     does not determine an exact match for one or more keys, but the possible
--- 188,232 ----
    list of heap pointers (PL, posting list) if the list is small enough.
   </para>
  
+  <sect2 id="gin-fast-update">
+   <title>GIN fast update technique</title>
+ 
+   <para>
+    Updating a <acronym>GIN</acronym> index tends to be slow because of the
+    intrinsic nature of inverted indexes: inserting or updating one heap row
+    can cause many inserts into the index (one for each key extracted
+    from the indexed value). As of
+    <productname>PostgreSQL</productname> 8.4, this problem is alleviated
+    by postponing most of the work until the next <command>VACUUM</>.
+    Newly inserted index entries are temporarily stored in an unsorted list of
+    pending entries.  <command>VACUUM</> inserts all pending entries into the
+    main <acronym>GIN</acronym> index data structure,
+    using the same bulk insert techniques used during initial index creation.
+    This greatly improves <acronym>GIN</acronym> index update speed, even
+    counting the additional vacuum overhead.
+   </para>
+ 
+   <para>
+    The disadvantage of this approach is that searches must scan the list
+    of pending entries in addition to searching the regular index, and so
+    a large list of pending entries will slow searches significantly.
+    It's recommended to use properly-configured autovacuum with tables
+    having <acronym>GIN</acronym> indexes, to keep this overhead to
+    reasonable levels.
+   </para>
+ 
+   <para>
+    If consistently-fast search speed is more important than update speed,
+    use of pending entries can be disabled by turning off the
+    <literal>FASTUPDATE</literal> storage parameter for a
+    <acronym>GIN</acronym> index.  See <xref linkend="sql-createindex"
+    endterm="sql-createindex-title"> for details.
+   </para>
+  </sect2>
+ 
   <sect2 id="gin-partial-match">
    <title>Partial match algorithm</title>
! 
    <para>
     GIN can support <quote>partial match</> queries, in which the query
     does not determine an exact match for one or more keys, but the possible
***************
*** 225,235 ****
     <term>Create vs insert</term>
     <listitem>
      <para>
!      In most cases, insertion into a <acronym>GIN</acronym> index is slow
       due to the likelihood of many keys being inserted for each value.
       So, for bulk insertions into a table it is advisable to drop the GIN
       index and recreate it after finishing bulk insertion.
      </para>
     </listitem>
    </varlistentry>
  
--- 261,278 ----
     <term>Create vs insert</term>
     <listitem>
      <para>
!      Insertion into a <acronym>GIN</acronym> index can be slow
       due to the likelihood of many keys being inserted for each value.
       So, for bulk insertions into a table it is advisable to drop the GIN
       index and recreate it after finishing bulk insertion.
      </para>
+ 
+     <para>
+      As of <productname>PostgreSQL</productname> 8.4, this advice is less
+      necessary since delayed indexing is used (see <xref
+      linkend="gin-fast-update"> for details).  But for very large updates
+      it may still be best to drop and recreate the index.
+     </para>
     </listitem>
    </varlistentry>
  
diff -c -r -N doc.orig/src/sgml/ref/create_index.sgml doc/src/sgml/ref/create_index.sgml
*** doc.orig/src/sgml/ref/create_index.sgml	2009-02-04 18:13:39.000000000 +0300
--- doc/src/sgml/ref/create_index.sgml	2009-02-04 18:23:23.000000000 +0300
***************
*** 294,299 ****
--- 294,329 ----
  
     </variablelist>
  
+    <para>
+     <acronym>GIN</acronym> indexes accept an additional parameter:
+    </para>
+ 
+    <variablelist>
+ 
+    <varlistentry>
+     <term><literal>FASTUPDATE</></term>
+     <listitem>
+     <para>
+      This setting controls usage of the fast update technique described in
+      <xref linkend="gin-fast-update">.  It is a Boolean parameter:
+      <literal>ON</> enables fast update, <literal>OFF</> disables it.
+      (Alternative spellings of <literal>ON</> and <literal>OFF</> are
+      allowed as described in <xref linkend="config-setting">.)  The
+      default is <literal>ON</>.
+     </para>
+ 
+     <note>
+      <para>
+       Turning <literal>FASTUPDATE</> off via <command>ALTER INDEX</> prevents
+       future insertions from going into the list of pending index entries,
+       but does not in itself flush previous entries.  You might want to do a
+       <command>VACUUM</> afterward to ensure the pending list is emptied.
+      </para>
+     </note>
+     </listitem>
+    </varlistentry>
+ 
+    </variablelist>
    </refsect2>
  
    <refsect2 id="SQL-CREATEINDEX-CONCURRENTLY">
***************
*** 502,507 ****
--- 532,544 ----
    </para>
  
    <para>
+    To create a <acronym>GIN</> index with fast update turned off:
+ <programlisting>
+ CREATE INDEX gin_idx ON documents_table (locations) WITH (fastupdate = off);
+ </programlisting>
+   </para>
+ 
+   <para>
     To create an index on the column <literal>code</> in the table
     <literal>films</> and have the index reside in the tablespace
     <literal>indexspace</>:
diff -c -r -N doc.orig/src/sgml/ref/vacuum.sgml doc/src/sgml/ref/vacuum.sgml
*** doc.orig/src/sgml/ref/vacuum.sgml	2009-02-04 18:13:39.000000000 +0300
--- doc/src/sgml/ref/vacuum.sgml	2009-02-04 18:23:23.000000000 +0300
***************
*** 63,68 ****
--- 63,75 ----
     blocks.  This form is much slower and requires an exclusive lock on each
     table while it is being processed.
    </para>
+ 
+   <para>
+    For tables with <acronym>GIN</> indexes, <command>VACUUM</command> (in
+    any form) also completes any delayed index insertions, by moving pending
+    index entries to the appropriate places in the main <acronym>GIN</> index
+    structure.  (See <xref linkend="gin-fast-update"> for more details.)
+   </para>
   </refsect1>
  
   <refsect1>
diff -c -r -N doc.orig/src/sgml/textsearch.sgml doc/src/sgml/textsearch.sgml
*** doc.orig/src/sgml/textsearch.sgml	2009-02-04 18:13:39.000000000 +0300
--- doc/src/sgml/textsearch.sgml	2009-02-04 18:23:23.000000000 +0300
***************
*** 3224,3230 ****
      </listitem>
      <listitem>
       <para>
!       GIN indexes are about ten times slower to update than GiST
       </para>
      </listitem>
      <listitem>
--- 3224,3232 ----
      </listitem>
      <listitem>
       <para>
!       GIN indexes are moderately slower to update than GiST indexes, but
!       about 10 times slower if fast update support was disabled
!       (see <xref linkend="gin-fast-update"> for details)
       </para>
      </listitem>
      <listitem>
