nbtxlog.c

PostgreSQL7.4.6 for Linux

			elog(PANIC, "btree_delete_page_redo: uninitialized right sibling");
		if (XLByteLE(lsn, PageGetLSN(page)))
			UnlockAndReleaseBuffer(buffer);
		else
		{
			pageop = (BTPageOpaque) PageGetSpecialPointer(page);
			pageop->btpo_prev = leftsib;

			PageSetLSN(page, lsn);
			PageSetSUI(page, ThisStartUpID);
			UnlockAndWriteBuffer(buffer);
		}
	}

	/* Fix right-link of left sibling, if any */
	if (redo && !(record->xl_info & XLR_BKP_BLOCK_3))
	{
		if (leftsib != P_NONE)
		{
			buffer = XLogReadBuffer(false, reln, leftsib);
			if (!BufferIsValid(buffer))
				elog(PANIC, "btree_delete_page_redo: lost left sibling");
			page = (Page) BufferGetPage(buffer);
			if (PageIsNew((PageHeader) page))
				elog(PANIC, "btree_delete_page_redo: uninitialized left sibling");
			if (XLByteLE(lsn, PageGetLSN(page)))
				UnlockAndReleaseBuffer(buffer);
			else
			{
				pageop = (BTPageOpaque) PageGetSpecialPointer(page);
				pageop->btpo_next = rightsib;

				PageSetLSN(page, lsn);
				PageSetSUI(page, ThisStartUpID);
				UnlockAndWriteBuffer(buffer);
			}
		}
	}

	/* Rewrite target page as empty deleted page */
	buffer = XLogReadBuffer(false, reln, target);
	if (!BufferIsValid(buffer))
		elog(PANIC, "btree_delete_page_%s: lost target page", op);
	page = (Page) BufferGetPage(buffer);
	if (redo)
		_bt_pageinit(page, BufferGetPageSize(buffer));
	else if (PageIsNew((PageHeader) page))
		elog(PANIC, "btree_delete_page_undo: uninitialized target page");
	pageop = (BTPageOpaque) PageGetSpecialPointer(page);

	if (redo)
	{
		pageop->btpo_prev = leftsib;
		pageop->btpo_next = rightsib;
		pageop->btpo.xact = FrozenTransactionId;
		pageop->btpo_flags = BTP_DELETED;

		PageSetLSN(page, lsn);
		PageSetSUI(page, ThisStartUpID);
		UnlockAndWriteBuffer(buffer);
	}
	else
	{
		/* undo */
		if (XLByteLT(PageGetLSN(page), lsn))
			elog(PANIC, "btree_delete_page_undo: bad left sibling LSN");
		elog(PANIC, "btree_delete_page_undo: unimplemented");
	}

	/* Update metapage if needed */
	if (redo)					/* metapage changes not undoable */
	{
		if (ismeta)
		{
			xl_btree_metadata md;

			memcpy(&md, (char *) xlrec + SizeOfBtreeDeletePage,
				   sizeof(xl_btree_metadata));
			_bt_restore_meta(reln, lsn,
							 md.root, md.level,
							 md.fastroot, md.fastlevel,
							 true);
		}
	}
}

static void
btree_xlog_newroot(bool redo, XLogRecPtr lsn, XLogRecord *record)
{
	xl_btree_newroot *xlrec = (xl_btree_newroot *) XLogRecGetData(record);
	Relation	reln;
	Buffer		buffer;
	Page		page;
	BTPageOpaque pageop;

	if (!redo)
		return;					/* not undoable */

	reln = XLogOpenRelation(redo, RM_BTREE_ID, xlrec->node);
	if (!RelationIsValid(reln))
		return;
	buffer = XLogReadBuffer(true, reln, xlrec->rootblk);
	if (!BufferIsValid(buffer))
		elog(PANIC, "btree_newroot_redo: no root page");

	page = (Page) BufferGetPage(buffer);
	_bt_pageinit(page, BufferGetPageSize(buffer));
	pageop = (BTPageOpaque) PageGetSpecialPointer(page);

	pageop->btpo_flags = BTP_ROOT;
	pageop->btpo_prev = pageop->btpo_next = P_NONE;
	pageop->btpo.level = xlrec->level;
	if (xlrec->level == 0)
		pageop->btpo_flags |= BTP_LEAF;

	if (record->xl_len > SizeOfBtreeNewroot)
		_bt_restore_page(page,
						 (char *) xlrec + SizeOfBtreeNewroot,
						 record->xl_len - SizeOfBtreeNewroot);

	PageSetLSN(page, lsn);
	PageSetSUI(page, ThisStartUpID);
	UnlockAndWriteBuffer(buffer);

	_bt_restore_meta(reln, lsn,
					 xlrec->rootblk, xlrec->level,
					 xlrec->rootblk, xlrec->level,
					 true);

	/* Check to see if this satisfies any incomplete insertions */
	if (record->xl_len > SizeOfBtreeNewroot &&
		incomplete_splits != NIL)
	{
		forget_matching_split(reln, xlrec->node,
							  xlrec->rootblk,
							  P_FIRSTKEY,
							  true);
	}
}

static void
btree_xlog_newmeta(bool redo, XLogRecPtr lsn, XLogRecord *record,
				   bool markvalid)
{
	xl_btree_newmeta *xlrec = (xl_btree_newmeta *) XLogRecGetData(record);
	Relation	reln;

	if (!redo)
		return;					/* not undoable */

	reln = XLogOpenRelation(redo, RM_BTREE_ID, xlrec->node);
	if (!RelationIsValid(reln))
		return;

	_bt_restore_meta(reln, lsn,
					 xlrec->meta.root, xlrec->meta.level,
					 xlrec->meta.fastroot, xlrec->meta.fastlevel,
					 markvalid);
}

static void
btree_xlog_newpage(bool redo, XLogRecPtr lsn, XLogRecord *record)
{
	xl_btree_newpage *xlrec = (xl_btree_newpage *) XLogRecGetData(record);
	Relation	reln;
	Buffer		buffer;
	Page		page;

	if (!redo || (record->xl_info & XLR_BKP_BLOCK_1))
		return;

	reln = XLogOpenRelation(redo, RM_BTREE_ID, xlrec->node);
	if (!RelationIsValid(reln))
		return;
	buffer = XLogReadBuffer(true, reln, xlrec->blkno);
	if (!BufferIsValid(buffer))
		elog(PANIC, "btree_newpage_redo: block unfound");
	page = (Page) BufferGetPage(buffer);

	Assert(record->xl_len == SizeOfBtreeNewpage + BLCKSZ);
	memcpy(page, (char *) xlrec + SizeOfBtreeNewpage, BLCKSZ);

	PageSetLSN(page, lsn);
	PageSetSUI(page, ThisStartUpID);
	UnlockAndWriteBuffer(buffer);
}


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

	switch (info)
	{
		case XLOG_BTREE_INSERT_LEAF:
			btree_xlog_insert(true, true, false, lsn, record);
			break;
		case XLOG_BTREE_INSERT_UPPER:
			btree_xlog_insert(true, false, false, lsn, record);
			break;
		case XLOG_BTREE_INSERT_META:
			btree_xlog_insert(true, false, true, lsn, record);
			break;
		case XLOG_BTREE_SPLIT_L:
			btree_xlog_split(true, true, false, lsn, record);
			break;
		case XLOG_BTREE_SPLIT_R:
			btree_xlog_split(true, false, false, lsn, record);
			break;
		case XLOG_BTREE_SPLIT_L_ROOT:
			btree_xlog_split(true, true, true, lsn, record);
			break;
		case XLOG_BTREE_SPLIT_R_ROOT:
			btree_xlog_split(true, false, true, lsn, record);
			break;
		case XLOG_BTREE_DELETE:
			btree_xlog_delete(true, lsn, record);
			break;
		case XLOG_BTREE_DELETE_PAGE:
			btree_xlog_delete_page(true, false, lsn, record);
			break;
		case XLOG_BTREE_DELETE_PAGE_META:
			btree_xlog_delete_page(true, true, lsn, record);
			break;
		case XLOG_BTREE_NEWROOT:
			btree_xlog_newroot(true, lsn, record);
			break;
		case XLOG_BTREE_NEWMETA:
			btree_xlog_newmeta(true, lsn, record, true);
			break;
		case XLOG_BTREE_NEWPAGE:
			btree_xlog_newpage(true, lsn, record);
			break;
		case XLOG_BTREE_INVALIDMETA:
			btree_xlog_newmeta(true, lsn, record, false);
			break;
		default:
			elog(PANIC, "btree_redo: unknown op code %u", info);
	}
}

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

	switch (info)
	{
		case XLOG_BTREE_INSERT_LEAF:
			btree_xlog_insert(false, true, false, lsn, record);
			break;
		case XLOG_BTREE_INSERT_UPPER:
			btree_xlog_insert(false, false, false, lsn, record);
			break;
		case XLOG_BTREE_INSERT_META:
			btree_xlog_insert(false, false, true, lsn, record);
			break;
		case XLOG_BTREE_SPLIT_L:
			btree_xlog_split(false, true, false, lsn, record);
			break;
		case XLOG_BTREE_SPLIT_R:
			btree_xlog_split(false, false, false, lsn, record);
			break;
		case XLOG_BTREE_SPLIT_L_ROOT:
			btree_xlog_split(false, true, true, lsn, record);
			break;
		case XLOG_BTREE_SPLIT_R_ROOT:
			btree_xlog_split(false, false, true, lsn, record);
			break;
		case XLOG_BTREE_DELETE:
			btree_xlog_delete(false, lsn, record);
			break;
		case XLOG_BTREE_DELETE_PAGE:
			btree_xlog_delete_page(false, false, lsn, record);
			break;
		case XLOG_BTREE_DELETE_PAGE_META:
			btree_xlog_delete_page(false, true, lsn, record);
			break;
		case XLOG_BTREE_NEWROOT:
			btree_xlog_newroot(false, lsn, record);
			break;
		case XLOG_BTREE_NEWMETA:
			btree_xlog_newmeta(false, lsn, record, true);
			break;
		case XLOG_BTREE_NEWPAGE:
			btree_xlog_newpage(false, lsn, record);
			break;
		case XLOG_BTREE_INVALIDMETA:
			btree_xlog_newmeta(false, lsn, record, false);
			break;
		default:
			elog(PANIC, "btree_undo: unknown op code %u", info);
	}
}

static void
out_target(char *buf, xl_btreetid *target)
{
	sprintf(buf + strlen(buf), "node %u/%u; tid %u/%u",
			target->node.tblNode, target->node.relNode,
			ItemPointerGetBlockNumber(&(target->tid)),
			ItemPointerGetOffsetNumber(&(target->tid)));
}

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

	switch (info)
	{
		case XLOG_BTREE_INSERT_LEAF:
			{
				xl_btree_insert *xlrec = (xl_btree_insert *) rec;

				strcat(buf, "insert: ");
				out_target(buf, &(xlrec->target));
				break;
			}
		case XLOG_BTREE_INSERT_UPPER:
			{
				xl_btree_insert *xlrec = (xl_btree_insert *) rec;

				strcat(buf, "insert_upper: ");
				out_target(buf, &(xlrec->target));
				break;
			}
		case XLOG_BTREE_INSERT_META:
			{
				xl_btree_insert *xlrec = (xl_btree_insert *) rec;

				strcat(buf, "insert_meta: ");
				out_target(buf, &(xlrec->target));
				break;
			}
		case XLOG_BTREE_SPLIT_L:
			{
				xl_btree_split *xlrec = (xl_btree_split *) rec;

				strcat(buf, "split_l: ");
				out_target(buf, &(xlrec->target));
				sprintf(buf + strlen(buf), "; oth %u; rgh %u",
						xlrec->otherblk, xlrec->rightblk);
				break;
			}
		case XLOG_BTREE_SPLIT_R:
			{
				xl_btree_split *xlrec = (xl_btree_split *) rec;

				strcat(buf, "split_r: ");
				out_target(buf, &(xlrec->target));
				sprintf(buf + strlen(buf), "; oth %u; rgh %u",
						xlrec->otherblk, xlrec->rightblk);
				break;
			}
		case XLOG_BTREE_SPLIT_L_ROOT:
			{
				xl_btree_split *xlrec = (xl_btree_split *) rec;

				strcat(buf, "split_l_root: ");
				out_target(buf, &(xlrec->target));
				sprintf(buf + strlen(buf), "; oth %u; rgh %u",
						xlrec->otherblk, xlrec->rightblk);
				break;
			}
		case XLOG_BTREE_SPLIT_R_ROOT:
			{
				xl_btree_split *xlrec = (xl_btree_split *) rec;

				strcat(buf, "split_r_root: ");
				out_target(buf, &(xlrec->target));
				sprintf(buf + strlen(buf), "; oth %u; rgh %u",
						xlrec->otherblk, xlrec->rightblk);
				break;
			}
		case XLOG_BTREE_DELETE:
			{
				xl_btree_delete *xlrec = (xl_btree_delete *) rec;

				sprintf(buf + strlen(buf), "delete: node %u/%u; blk %u",
				 xlrec->node.tblNode, xlrec->node.relNode, xlrec->block);
				break;
			}
		case XLOG_BTREE_DELETE_PAGE:
		case XLOG_BTREE_DELETE_PAGE_META:
			{
				xl_btree_delete_page *xlrec = (xl_btree_delete_page *) rec;

				strcat(buf, "delete_page: ");
				out_target(buf, &(xlrec->target));
				sprintf(buf + strlen(buf), "; dead %u; left %u; right %u",
						xlrec->deadblk, xlrec->leftblk, xlrec->rightblk);
				break;
			}
		case XLOG_BTREE_NEWROOT:
			{
				xl_btree_newroot *xlrec = (xl_btree_newroot *) rec;

				sprintf(buf + strlen(buf), "newroot: node %u/%u; root %u lev %u",
						xlrec->node.tblNode, xlrec->node.relNode,
						xlrec->rootblk, xlrec->level);
				break;
			}
		case XLOG_BTREE_NEWMETA:
			{
				xl_btree_newmeta *xlrec = (xl_btree_newmeta *) rec;

				sprintf(buf + strlen(buf), "newmeta: node %u/%u; root %u lev %u fast %u lev %u",
						xlrec->node.tblNode, xlrec->node.relNode,
						xlrec->meta.root, xlrec->meta.level,
						xlrec->meta.fastroot, xlrec->meta.fastlevel);
				break;
			}
		case XLOG_BTREE_NEWPAGE:
			{
				xl_btree_newpage *xlrec = (xl_btree_newpage *) rec;

				sprintf(buf + strlen(buf), "newpage: node %u/%u; page %u",
						xlrec->node.tblNode, xlrec->node.relNode,
						xlrec->blkno);
				break;
			}
		case XLOG_BTREE_INVALIDMETA:
			{
				xl_btree_newmeta *xlrec = (xl_btree_newmeta *) rec;

				sprintf(buf + strlen(buf), "invalidmeta: node %u/%u; root %u lev %u fast %u lev %u",
						xlrec->node.tblNode, xlrec->node.relNode,
						xlrec->meta.root, xlrec->meta.level,
						xlrec->meta.fastroot, xlrec->meta.fastlevel);
				break;
			}
		default:
			strcat(buf, "UNKNOWN");
			break;
	}
}

void
btree_xlog_startup(void)
{
	incomplete_splits = NIL;
}

void
btree_xlog_cleanup(void)
{
	List	   *l;

	foreach(l, incomplete_splits)
	{
		bt_incomplete_split *split = (bt_incomplete_split *) lfirst(l);
		Relation	reln;
		Buffer		lbuf,
					rbuf;
		Page		lpage,
					rpage;
		BTPageOpaque lpageop,
					rpageop;
		bool		is_only;

		reln = XLogOpenRelation(true, RM_BTREE_ID, split->node);
		if (!RelationIsValid(reln))
			continue;
		lbuf = XLogReadBuffer(false, reln, split->leftblk);
		if (!BufferIsValid(lbuf))
			elog(PANIC, "btree_xlog_cleanup: left block unfound");
		lpage = (Page) BufferGetPage(lbuf);
		lpageop = (BTPageOpaque) PageGetSpecialPointer(lpage);
		rbuf = XLogReadBuffer(false, reln, split->rightblk);
		if (!BufferIsValid(rbuf))
			elog(PANIC, "btree_xlog_cleanup: right block unfound");
		rpage = (Page) BufferGetPage(rbuf);
		rpageop = (BTPageOpaque) PageGetSpecialPointer(rpage);

		/* if the two pages are all of their level, it's a only-page split */
		is_only = P_LEFTMOST(lpageop) && P_RIGHTMOST(rpageop);

		_bt_insert_parent(reln, lbuf, rbuf, (BTStack) NULL,
						  split->is_root, is_only);
	}
	incomplete_splits = NIL;
}