gistxlog.c

postgresql8.3.4源码,开源数据库

/*-------------------------------------------------------------------------
 *
 * gistxlog.c
 *	  WAL replay logic for GiST.
 *
 *
 * Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *			 $PostgreSQL: pgsql/src/backend/access/gist/gistxlog.c,v 1.27 2008/01/01 19:45:46 momjian Exp $
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/gist_private.h"
#include "access/heapam.h"
#include "miscadmin.h"
#include "utils/memutils.h"


typedef struct
{
	gistxlogPageUpdate *data;
	int			len;
	IndexTuple *itup;
	OffsetNumber *todelete;
} PageUpdateRecord;

typedef struct
{
	gistxlogPage *header;
	IndexTuple *itup;
} NewPage;

typedef struct
{
	gistxlogPageSplit *data;
	NewPage    *page;
} PageSplitRecord;

/* track for incomplete inserts, idea was taken from nbtxlog.c */

typedef struct gistIncompleteInsert
{
	RelFileNode node;
	BlockNumber origblkno;		/* for splits */
	ItemPointerData key;
	int			lenblk;
	BlockNumber *blkno;
	XLogRecPtr	lsn;
	BlockNumber *path;
	int			pathlen;
} gistIncompleteInsert;


static MemoryContext opCtx;		/* working memory for operations */
static MemoryContext insertCtx; /* holds incomplete_inserts list */
static List *incomplete_inserts;


#define ItemPointerEQ(a, b) \
	( ItemPointerGetOffsetNumber(a) == ItemPointerGetOffsetNumber(b) && \
	  ItemPointerGetBlockNumber (a) == ItemPointerGetBlockNumber(b) )


static void
pushIncompleteInsert(RelFileNode node, XLogRecPtr lsn, ItemPointerData key,
					 BlockNumber *blkno, int lenblk,
					 PageSplitRecord *xlinfo /* to extract blkno info */ )
{
	MemoryContext oldCxt;
	gistIncompleteInsert *ninsert;

	if (!ItemPointerIsValid(&key))

		/*
		 * if key is null then we should not store insertion as incomplete,
		 * because it's a vacuum operation..
		 */
		return;

	oldCxt = MemoryContextSwitchTo(insertCtx);
	ninsert = (gistIncompleteInsert *) palloc(sizeof(gistIncompleteInsert));

	ninsert->node = node;
	ninsert->key = key;
	ninsert->lsn = lsn;

	if (lenblk && blkno)
	{
		ninsert->lenblk = lenblk;
		ninsert->blkno = (BlockNumber *) palloc(sizeof(BlockNumber) * ninsert->lenblk);
		memcpy(ninsert->blkno, blkno, sizeof(BlockNumber) * ninsert->lenblk);
		ninsert->origblkno = *blkno;
	}
	else
	{
		int			i;

		Assert(xlinfo);
		ninsert->lenblk = xlinfo->data->npage;
		ninsert->blkno = (BlockNumber *) palloc(sizeof(BlockNumber) * ninsert->lenblk);
		for (i = 0; i < ninsert->lenblk; i++)
			ninsert->blkno[i] = xlinfo->page[i].header->blkno;
		ninsert->origblkno = xlinfo->data->origblkno;
	}
	Assert(ninsert->lenblk > 0);

	/*
	 * Stick the new incomplete insert onto the front of the list, not the
	 * back.  This is so that gist_xlog_cleanup will process incompletions in
	 * last-in-first-out order.
	 */
	incomplete_inserts = lcons(ninsert, incomplete_inserts);

	MemoryContextSwitchTo(oldCxt);
}

static void
forgetIncompleteInsert(RelFileNode node, ItemPointerData key)
{
	ListCell   *l;

	if (!ItemPointerIsValid(&key))
		return;

	if (incomplete_inserts == NIL)
		return;

	foreach(l, incomplete_inserts)
	{
		gistIncompleteInsert *insert = (gistIncompleteInsert *) lfirst(l);

		if (RelFileNodeEquals(node, insert->node) && ItemPointerEQ(&(insert->key), &(key)))
		{
			/* found */
			incomplete_inserts = list_delete_ptr(incomplete_inserts, insert);
			pfree(insert->blkno);
			pfree(insert);
			break;
		}
	}
}

static void
decodePageUpdateRecord(PageUpdateRecord *decoded, XLogRecord *record)
{
	char	   *begin = XLogRecGetData(record),
			   *ptr;
	int			i = 0,
				addpath = 0;

	decoded->data = (gistxlogPageUpdate *) begin;

	if (decoded->data->ntodelete)
	{
		decoded->todelete = (OffsetNumber *) (begin + sizeof(gistxlogPageUpdate) + addpath);
		addpath = MAXALIGN(sizeof(OffsetNumber) * decoded->data->ntodelete);
	}
	else
		decoded->todelete = NULL;

	decoded->len = 0;
	ptr = begin + sizeof(gistxlogPageUpdate) + addpath;
	while (ptr - begin < record->xl_len)
	{
		decoded->len++;
		ptr += IndexTupleSize((IndexTuple) ptr);
	}

	decoded->itup = (IndexTuple *) palloc(sizeof(IndexTuple) * decoded->len);

	ptr = begin + sizeof(gistxlogPageUpdate) + addpath;
	while (ptr - begin < record->xl_len)
	{
		decoded->itup[i] = (IndexTuple) ptr;
		ptr += IndexTupleSize(decoded->itup[i]);
		i++;
	}
}

/*
 * redo any page update (except page split)
 */
static void
gistRedoPageUpdateRecord(XLogRecPtr lsn, XLogRecord *record, bool isnewroot)
{
	gistxlogPageUpdate *xldata = (gistxlogPageUpdate *) XLogRecGetData(record);
	PageUpdateRecord xlrec;
	Relation	reln;
	Buffer		buffer;
	Page		page;

	/* we must fix incomplete_inserts list even if XLR_BKP_BLOCK_1 is set */
	forgetIncompleteInsert(xldata->node, xldata->key);

	if (!isnewroot && xldata->blkno != GIST_ROOT_BLKNO)
		/* operation with root always finalizes insertion */
		pushIncompleteInsert(xldata->node, lsn, xldata->key,
							 &(xldata->blkno), 1,
							 NULL);

	/* nothing else to do if page was backed up (and no info to do it with) */
	if (record->xl_info & XLR_BKP_BLOCK_1)
		return;

	decodePageUpdateRecord(&xlrec, record);

	reln = XLogOpenRelation(xlrec.data->node);
	buffer = XLogReadBuffer(reln, xlrec.data->blkno, false);
	if (!BufferIsValid(buffer))
		return;
	page = (Page) BufferGetPage(buffer);

	if (XLByteLE(lsn, PageGetLSN(page)))
	{
		UnlockReleaseBuffer(buffer);
		return;
	}

	if (isnewroot)
		GISTInitBuffer(buffer, 0);
	else if (xlrec.data->ntodelete)
	{
		int			i;

		for (i = 0; i < xlrec.data->ntodelete; i++)
			PageIndexTupleDelete(page, xlrec.todelete[i]);
		if (GistPageIsLeaf(page))
			GistMarkTuplesDeleted(page);
	}

	/* add tuples */
	if (xlrec.len > 0)
		gistfillbuffer(reln, page, xlrec.itup, xlrec.len, InvalidOffsetNumber);

	/*
	 * special case: leafpage, nothing to insert, nothing to delete, then
	 * vacuum marks page
	 */
	if (GistPageIsLeaf(page) && xlrec.len == 0 && xlrec.data->ntodelete == 0)
		GistClearTuplesDeleted(page);

	if (!GistPageIsLeaf(page) && PageGetMaxOffsetNumber(page) == InvalidOffsetNumber && xldata->blkno == GIST_ROOT_BLKNO)

		/*
		 * all links on non-leaf root page was deleted by vacuum full, so root
		 * page becomes a leaf
		 */
		GistPageSetLeaf(page);

	GistPageGetOpaque(page)->rightlink = InvalidBlockNumber;
	PageSetLSN(page, lsn);
	PageSetTLI(page, ThisTimeLineID);
	MarkBufferDirty(buffer);
	UnlockReleaseBuffer(buffer);
}

static void
gistRedoPageDeleteRecord(XLogRecPtr lsn, XLogRecord *record)
{
	gistxlogPageDelete *xldata = (gistxlogPageDelete *) XLogRecGetData(record);
	Relation	reln;
	Buffer		buffer;
	Page		page;

	/* nothing else to do if page was backed up (and no info to do it with) */
	if (record->xl_info & XLR_BKP_BLOCK_1)
		return;

	reln = XLogOpenRelation(xldata->node);
	buffer = XLogReadBuffer(reln, xldata->blkno, false);
	if (!BufferIsValid(buffer))
		return;

	page = (Page) BufferGetPage(buffer);
	GistPageSetDeleted(page);

	PageSetLSN(page, lsn);
	PageSetTLI(page, ThisTimeLineID);
	MarkBufferDirty(buffer);
	UnlockReleaseBuffer(buffer);
}

static void
decodePageSplitRecord(PageSplitRecord *decoded, XLogRecord *record)
{
	char	   *begin = XLogRecGetData(record),
			   *ptr;
	int			j,
				i = 0;

	decoded->data = (gistxlogPageSplit *) begin;
	decoded->page = (NewPage *) palloc(sizeof(NewPage) * decoded->data->npage);

	ptr = begin + sizeof(gistxlogPageSplit);
	for (i = 0; i < decoded->data->npage; i++)
	{
		Assert(ptr - begin < record->xl_len);
		decoded->page[i].header = (gistxlogPage *) ptr;
		ptr += sizeof(gistxlogPage);

		decoded->page[i].itup = (IndexTuple *)
			palloc(sizeof(IndexTuple) * decoded->page[i].header->num);
		j = 0;
		while (j < decoded->page[i].header->num)
		{
			Assert(ptr - begin < record->xl_len);
			decoded->page[i].itup[j] = (IndexTuple) ptr;
			ptr += IndexTupleSize((IndexTuple) ptr);
			j++;
		}
	}
}

static void
gistRedoPageSplitRecord(XLogRecPtr lsn, XLogRecord *record)
{
	PageSplitRecord xlrec;
	Relation	reln;
	Buffer		buffer;
	Page		page;
	int			i;
	int			flags;

	decodePageSplitRecord(&xlrec, record);
	reln = XLogOpenRelation(xlrec.data->node);
	flags = xlrec.data->origleaf ? F_LEAF : 0;

	/* loop around all pages */
	for (i = 0; i < xlrec.data->npage; i++)
	{
		NewPage    *newpage = xlrec.page + i;

		buffer = XLogReadBuffer(reln, newpage->header->blkno, true);
		Assert(BufferIsValid(buffer));
		page = (Page) BufferGetPage(buffer);

		/* ok, clear buffer */
		GISTInitBuffer(buffer, flags);

		/* and fill it */
		gistfillbuffer(reln, page, newpage->itup, newpage->header->num, FirstOffsetNumber);

		PageSetLSN(page, lsn);
		PageSetTLI(page, ThisTimeLineID);
		MarkBufferDirty(buffer);
		UnlockReleaseBuffer(buffer);
	}

	forgetIncompleteInsert(xlrec.data->node, xlrec.data->key);

	pushIncompleteInsert(xlrec.data->node, lsn, xlrec.data->key,
						 NULL, 0,
						 &xlrec);
}

static void
gistRedoCreateIndex(XLogRecPtr lsn, XLogRecord *record)
{
	RelFileNode *node = (RelFileNode *) XLogRecGetData(record);
	Relation	reln;
	Buffer		buffer;
	Page		page;

	reln = XLogOpenRelation(*node);
	buffer = XLogReadBuffer(reln, GIST_ROOT_BLKNO, true);
	Assert(BufferIsValid(buffer));
	page = (Page) BufferGetPage(buffer);

	GISTInitBuffer(buffer, F_LEAF);

	PageSetLSN(page, lsn);
	PageSetTLI(page, ThisTimeLineID);

	MarkBufferDirty(buffer);
	UnlockReleaseBuffer(buffer);
}

static void
gistRedoCompleteInsert(XLogRecPtr lsn, XLogRecord *record)
{
	char	   *begin = XLogRecGetData(record),
			   *ptr;
	gistxlogInsertComplete *xlrec;

	xlrec = (gistxlogInsertComplete *) begin;

	ptr = begin + sizeof(gistxlogInsertComplete);
	while (ptr - begin < record->xl_len)
	{
		Assert(record->xl_len - (ptr - begin) >= sizeof(ItemPointerData));
		forgetIncompleteInsert(xlrec->node, *((ItemPointerData *) ptr));
		ptr += sizeof(ItemPointerData);
	}
}

void
gist_redo(XLogRecPtr lsn, XLogRecord *record)
{
	uint8		info = record->xl_info & ~XLR_INFO_MASK;

	MemoryContext oldCxt;

	oldCxt = MemoryContextSwitchTo(opCtx);
	switch (info)
	{
		case XLOG_GIST_PAGE_UPDATE:
			gistRedoPageUpdateRecord(lsn, record, false);
			break;
		case XLOG_GIST_PAGE_DELETE:
			gistRedoPageDeleteRecord(lsn, record);
			break;
		case XLOG_GIST_NEW_ROOT:
			gistRedoPageUpdateRecord(lsn, record, true);
			break;
		case XLOG_GIST_PAGE_SPLIT:
			gistRedoPageSplitRecord(lsn, record);
			break;
		case XLOG_GIST_CREATE_INDEX:
			gistRedoCreateIndex(lsn, record);
			break;
		case XLOG_GIST_INSERT_COMPLETE:
			gistRedoCompleteInsert(lsn, record);
			break;
		default:
			elog(PANIC, "gist_redo: unknown op code %u", info);
	}

	MemoryContextSwitchTo(oldCxt);
	MemoryContextReset(opCtx);
}

static void
out_target(StringInfo buf, RelFileNode node, ItemPointerData key)
{
	appendStringInfo(buf, "rel %u/%u/%u",
					 node.spcNode, node.dbNode, node.relNode);
	if (ItemPointerIsValid(&key))
		appendStringInfo(buf, "; tid %u/%u",
						 ItemPointerGetBlockNumber(&key),
						 ItemPointerGetOffsetNumber(&key));
}

static void
out_gistxlogPageUpdate(StringInfo buf, gistxlogPageUpdate *xlrec)
{
	out_target(buf, xlrec->node, xlrec->key);
	appendStringInfo(buf, "; block number %u", xlrec->blkno);
}

static void
out_gistxlogPageDelete(StringInfo buf, gistxlogPageDelete *xlrec)
{
	appendStringInfo(buf, "page_delete: rel %u/%u/%u; blkno %u",
				xlrec->node.spcNode, xlrec->node.dbNode, xlrec->node.relNode,
					 xlrec->blkno);
}

static void
out_gistxlogPageSplit(StringInfo buf, gistxlogPageSplit *xlrec)
{
	appendStringInfo(buf, "page_split: ");
	out_target(buf, xlrec->node, xlrec->key);
	appendStringInfo(buf, "; block number %u splits to %d pages",
					 xlrec->origblkno, xlrec->npage);
}

void
gist_desc(StringInfo buf, uint8 xl_info, char *rec)
{
	uint8		info = xl_info & ~XLR_INFO_MASK;

	switch (info)
	{
		case XLOG_GIST_PAGE_UPDATE:
			appendStringInfo(buf, "page_update: ");
			out_gistxlogPageUpdate(buf, (gistxlogPageUpdate *) rec);
			break;
		case XLOG_GIST_PAGE_DELETE:
			out_gistxlogPageDelete(buf, (gistxlogPageDelete *) rec);
			break;
		case XLOG_GIST_NEW_ROOT:
			appendStringInfo(buf, "new_root: ");
			out_target(buf, ((gistxlogPageUpdate *) rec)->node, ((gistxlogPageUpdate *) rec)->key);
			break;
		case XLOG_GIST_PAGE_SPLIT:
			out_gistxlogPageSplit(buf, (gistxlogPageSplit *) rec);
			break;
		case XLOG_GIST_CREATE_INDEX:
			appendStringInfo(buf, "create_index: rel %u/%u/%u",
							 ((RelFileNode *) rec)->spcNode,
							 ((RelFileNode *) rec)->dbNode,
							 ((RelFileNode *) rec)->relNode);
			break;