gistvacuum.c

PostgreSQL 8.1.4的源码 适用于Linux下的开源数据库系统

/*-------------------------------------------------------------------------
 *
 * gistvacuum.c
 *	  interface routines for the postgres GiST index access method.
 *
 *
 * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  $PostgreSQL: pgsql/src/backend/access/gist/gistvacuum.c,v 1.9.2.1 2006/02/14 16:39:36 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/genam.h"
#include "access/gist_private.h"
#include "access/gistscan.h"
#include "access/heapam.h"
#include "catalog/index.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "utils/memutils.h"
#include "storage/freespace.h"
#include "storage/smgr.h"

/* filled by gistbulkdelete, cleared by gistvacuumpcleanup */
static bool needFullVacuum = false;


typedef struct
{
	GISTSTATE	giststate;
	Relation	index;
	MemoryContext opCtx;
	IndexBulkDeleteResult *result;
} GistVacuum;

typedef struct
{
	IndexTuple *itup;
	int			ituplen;
	bool		emptypage;
} ArrayTuple;

static ArrayTuple
gistVacuumUpdate(GistVacuum *gv, BlockNumber blkno, bool needunion)
{
	ArrayTuple	res = {NULL, 0, false};
	Buffer		buffer;
	Page		page;
	OffsetNumber i,
				maxoff;
	ItemId		iid;
	int			lenaddon = 4,
				curlenaddon = 0,
				ntodelete = 0;
	IndexTuple	idxtuple,
			   *addon = NULL;
	bool		needwrite = false;
	OffsetNumber todelete[MaxOffsetNumber];
	ItemPointerData *completed = NULL;
	int			ncompleted = 0,
				lencompleted = 16;

	vacuum_delay_point();

	buffer = ReadBuffer(gv->index, blkno);
	page = (Page) BufferGetPage(buffer);
	maxoff = PageGetMaxOffsetNumber(page);

	if (GistPageIsLeaf(page))
	{
		if (GistTuplesDeleted(page))
		{
			needunion = needwrite = true;
			GistClearTuplesDeleted(page);
		}
	}
	else
	{
		completed = (ItemPointerData *) palloc(sizeof(ItemPointerData) * lencompleted);
		addon = (IndexTuple *) palloc(sizeof(IndexTuple) * lenaddon);

		for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
		{
			ArrayTuple	chldtuple;
			bool		needchildunion;

			iid = PageGetItemId(page, i);
			idxtuple = (IndexTuple) PageGetItem(page, iid);
			needchildunion = (GistTupleIsInvalid(idxtuple)) ? true : false;

			if (needchildunion)
				elog(DEBUG2, "gistVacuumUpdate: need union for block %u",
					 ItemPointerGetBlockNumber(&(idxtuple->t_tid)));

			chldtuple = gistVacuumUpdate(gv, ItemPointerGetBlockNumber(&(idxtuple->t_tid)),
										 needchildunion);
			if (chldtuple.ituplen || chldtuple.emptypage)
			{
				PageIndexTupleDelete(page, i);
				todelete[ntodelete++] = i;
				i--;
				maxoff--;
				needwrite = needunion = true;

				if (chldtuple.ituplen)
				{
					while (curlenaddon + chldtuple.ituplen >= lenaddon)
					{
						lenaddon *= 2;
						addon = (IndexTuple *) repalloc(addon, sizeof(IndexTuple) * lenaddon);
					}

					memcpy(addon + curlenaddon, chldtuple.itup, chldtuple.ituplen * sizeof(IndexTuple));

					curlenaddon += chldtuple.ituplen;

					if (chldtuple.ituplen > 1)
					{
						/*
						 * child was splitted, so we need mark completion
						 * insert(split)
						 */
						int			j;

						while (ncompleted + chldtuple.ituplen > lencompleted)
						{
							lencompleted *= 2;
							completed = (ItemPointerData *) repalloc(completed, sizeof(ItemPointerData) * lencompleted);
						}
						for (j = 0; j < chldtuple.ituplen; j++)
						{
							ItemPointerCopy(&(chldtuple.itup[j]->t_tid), completed + ncompleted);
							ncompleted++;
						}
					}
					pfree(chldtuple.itup);
				}
			}
		}

		if (curlenaddon)
		{
			/* insert updated tuples */
			if (gistnospace(page, addon, curlenaddon))
			{
				/* there is no space on page to insert tuples */
				IndexTuple *vec;
				SplitedPageLayout *dist = NULL,
						   *ptr;
				int			i;
				MemoryContext oldCtx = MemoryContextSwitchTo(gv->opCtx);

				vec = gistextractbuffer(buffer, &(res.ituplen));
				vec = gistjoinvector(vec, &(res.ituplen), addon, curlenaddon);
				res.itup = gistSplit(gv->index, buffer, vec, &(res.ituplen), &dist, &(gv->giststate));
				MemoryContextSwitchTo(oldCtx);

				vec = (IndexTuple *) palloc(sizeof(IndexTuple) * res.ituplen);
				for (i = 0; i < res.ituplen; i++)
				{
					vec[i] = (IndexTuple) palloc(IndexTupleSize(res.itup[i]));
					memcpy(vec[i], res.itup[i], IndexTupleSize(res.itup[i]));
				}
				res.itup = vec;

				if (!gv->index->rd_istemp)
				{
					XLogRecPtr	recptr;
					XLogRecData *rdata;
					ItemPointerData key;		/* set key for incomplete
												 * insert */
					char	   *xlinfo;

					ItemPointerSet(&key, blkno, TUPLE_IS_VALID);

					rdata = formSplitRdata(gv->index->rd_node, blkno,
										   &key, dist);
					xlinfo = rdata->data;

					START_CRIT_SECTION();

					recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_PAGE_SPLIT, rdata);
					ptr = dist;
					while (ptr)
					{
						PageSetLSN(BufferGetPage(ptr->buffer), recptr);
						PageSetTLI(BufferGetPage(ptr->buffer), ThisTimeLineID);
						ptr = ptr->next;
					}

					END_CRIT_SECTION();
					pfree(xlinfo);
					pfree(rdata);
				}
				else
				{
					ptr = dist;
					while (ptr)
					{
						PageSetLSN(BufferGetPage(ptr->buffer), XLogRecPtrForTemp);
						ptr = ptr->next;
					}
				}

				ptr = dist;
				while (ptr)
				{
					if (BufferGetBlockNumber(ptr->buffer) != blkno)
						LockBuffer(ptr->buffer, GIST_UNLOCK);
					WriteBuffer(ptr->buffer);
					ptr = ptr->next;
				}

				if (blkno == GIST_ROOT_BLKNO)
				{
					ItemPointerData key;		/* set key for incomplete
												 * insert */

					ItemPointerSet(&key, blkno, TUPLE_IS_VALID);

					oldCtx = MemoryContextSwitchTo(gv->opCtx);
					gistnewroot(gv->index, buffer, res.itup, res.ituplen, &key);
					MemoryContextSwitchTo(oldCtx);

					WriteNoReleaseBuffer(buffer);
				}

				needwrite = false;

				MemoryContextReset(gv->opCtx);

				needunion = false;		/* gistSplit already forms unions */
			}
			else
			{
				/* enough free space */
				gistfillbuffer(gv->index, page, addon, curlenaddon, InvalidOffsetNumber);
			}
		}
	}

	if (needunion)
	{
		/* forms union for page  or check empty */
		if (PageIsEmpty(page))
		{
			if (blkno == GIST_ROOT_BLKNO)
			{
				needwrite = true;
				GistPageSetLeaf(page);
			}
			else
			{
				needwrite = true;
				res.emptypage = true;
				GistPageSetDeleted(page);
				gv->result->pages_deleted++;
			}
		}
		else
		{
			IndexTuple *vec,
						tmp;
			int			veclen = 0;
			MemoryContext oldCtx = MemoryContextSwitchTo(gv->opCtx);

			vec = gistextractbuffer(buffer, &veclen);
			tmp = gistunion(gv->index, vec, veclen, &(gv->giststate));
			MemoryContextSwitchTo(oldCtx);

			res.itup = (IndexTuple *) palloc(sizeof(IndexTuple));
			res.ituplen = 1;
			res.itup[0] = (IndexTuple) palloc(IndexTupleSize(tmp));
			memcpy(res.itup[0], tmp, IndexTupleSize(tmp));

			ItemPointerSetBlockNumber(&(res.itup[0]->t_tid), blkno);
			GistTupleSetValid(res.itup[0]);

			MemoryContextReset(gv->opCtx);
		}
	}

	if (needwrite)
	{
		if (!gv->index->rd_istemp)
		{
			XLogRecData *rdata;
			XLogRecPtr	recptr;
			char	   *xlinfo;

			rdata = formUpdateRdata(gv->index->rd_node, blkno, todelete, ntodelete,
									res.emptypage, addon, curlenaddon, NULL);
			xlinfo = rdata->data;

			START_CRIT_SECTION();
			recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_ENTRY_UPDATE, rdata);
			PageSetLSN(page, recptr);
			PageSetTLI(page, ThisTimeLineID);
			END_CRIT_SECTION();

			pfree(xlinfo);
			pfree(rdata);
		}
		else
			PageSetLSN(page, XLogRecPtrForTemp);
		WriteBuffer(buffer);
	}
	else
		ReleaseBuffer(buffer);

	if (ncompleted && !gv->index->rd_istemp)
		gistxlogInsertCompletion(gv->index->rd_node, completed, ncompleted);

	for (i = 0; i < curlenaddon; i++)
		pfree(addon[i]);
	if (addon)