redo/undo support functions and cleanups.

1 files changed, 284 insertions, 4 deletions

diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c
index 8f44fabaa54..a507c39bc24 100644
--- a/src/backend/access/transam/xlogutils.c
+++ b/src/backend/access/transam/xlogutils.c
@@ -1,6 +1,6 @@
 /*-------------------------------------------------------------------------
  *
- * xlog.c
+ * xlogutils.c
  *
  *
  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
@@ -9,12 +9,26 @@
  *-------------------------------------------------------------------------
  */
 
+#ifdef XLOG
+
 #include "postgres.h"
 
 #include "access/xlog.h"
 #include "access/xact.h"
+#include "storage/bufpage.h"
+#include "storage/bufmgr.h"
+#include "storage/smgr.h"
+#include "access/htup.h"
+#include "access/xlogutils.h"
+#include "catalog/pg_database.h"
 
-#ifdef XLOG
+/*
+ * ---------------------------------------------------------------
+ *
+ * Index support functions
+ *
+ *----------------------------------------------------------------
+ */
 
 /*
  * Check if specified heap tuple was inserted by given
@@ -59,7 +73,8 @@ XLogIsOwnerOfTuple(RelFileNode hnode, ItemPointer iptr,
 
 	htup = (HeapTupleHeader) PageGetItem(page, lp);
 
-	if (PageGetSUI(page) != ThisStartUpID || htup->t_xmin != xid || htup->t_cmin != cid)
+	Assert(PageGetSUI(page) == ThisStartUpID);
+	if (htup->t_xmin != xid || htup->t_cmin != cid)
 	{
 		UnlockAndReleaseBuffer(buffer);
 		return(-1);
@@ -70,6 +85,8 @@ XLogIsOwnerOfTuple(RelFileNode hnode, ItemPointer iptr,
 }
 
 /*
+ * MUST BE CALLED ONLY ON RECOVERY.
+ *
  * Check if exists valid (inserted by not aborted xaction) heap tuple
  * for given item pointer
  */
@@ -98,6 +115,14 @@ XLogIsValidTuple(RelFileNode hnode, ItemPointer iptr)
 		UnlockAndReleaseBuffer(buffer);
 		return(false);
 	}
+
+	if (PageGetSUI(page) != ThisStartUpID)
+	{
+		Assert(PageGetSUI(page) < ThisStartUpID);
+		UnlockAndReleaseBuffer(buffer);
+		return(true);
+	}
+
 	lp = PageGetItemId(page, ItemPointerGetOffsetNumber(iptr));
 	if (!ItemIdIsUsed(lp) || ItemIdDeleted(lp))
 	{
@@ -107,7 +132,9 @@ XLogIsValidTuple(RelFileNode hnode, ItemPointer iptr)
 
 	htup = (HeapTupleHeader) PageGetItem(page, lp);
 
-	if (XLogIsAborted(PageGetSUI(page), htup->t_xmin))
+	/* MUST CHECK WASN'T TUPLE INSERTED IN PREV STARTUP */
+
+	if (XLogIsAborted(htup->t_xmin))
 	{
 		UnlockAndReleaseBuffer(buffer);
 		return(false);
@@ -117,4 +144,257 @@ XLogIsValidTuple(RelFileNode hnode, ItemPointer iptr)
 	return(true);
 }
 
+/*
+ * ---------------------------------------------------------------
+ *
+ * Transaction support functions for recovery
+ *
+ * On recovery we create tmp file to know what xactions were
+ * committed/aborted (2 bits per xaction).
+ *
+ *----------------------------------------------------------------
+ */
+
+bool
+XLogIsAborted(TransactionId xid)
+{
+	return(false);
+}
+
+bool
+XLogIsCommitted(TransactionId xid)
+{
+	return(true);
+}
+
+void
+XLogMarkAborted(TransactionId xid)
+{
+	return;
+}
+
+void
+XLogMarkCommitted(TransactionId xid)
+{
+	return;
+}
+
+
+/*
+ * ---------------------------------------------------------------
+ *
+ * Storage related support functions
+ *
+ *----------------------------------------------------------------
+ */
+
+Buffer
+XLogReadBuffer(bool extend, Relation reln, BlockNumber blkno)
+{
+	BlockNumber	lastblock = RelationGetNumberOfBlocks(reln);
+	Buffer		buffer;
+
+	if (blkno >= lastblock)
+	{
+		buffer = InvalidBuffer;
+		if (extend)		/* we do this in recovery only - no locks */
+		{
+			Assert(InRecovery);
+			while (lastblock <= blkno)
+			{
+				buffer = ReadBuffer(reln, P_NEW);
+				lastblock++;
+			}
+		}
+		if (buffer != InvalidBuffer)
+			LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		return(buffer);
+	}
+
+	buffer = ReadBuffer(reln, blkno);
+	if (buffer != InvalidBuffer)
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+	return(buffer);
+}
+
+/*
+ * "Relation" cache
+ */
+
+typedef struct XLogRelDesc
+{
+	RelationData			reldata;
+	struct XLogRelDesc	   *lessRecently;
+	struct XLogRelDesc	   *moreRecently;
+} XLogRelDesc;
+
+typedef struct XLogRelCacheEntry
+{
+	RelFileNode		rnode;
+	XLogRelDesc	   *rdesc;
+} XLogRelCacheEntry;
+
+static HTAB				   *_xlrelcache;
+static XLogRelDesc		   *_xlrelarr = NULL;
+static Form_pg_class		_xlpgcarr = NULL;
+static int					_xlast = 0;
+static int					_xlcnt = 0;
+#define	_XLOG_INITRELCACHESIZE	32
+#define	_XLOG_MAXRELCACHESIZE	512
+
+void
+XLogCloseRelationCache(void)
+{
+	int i;
+
+	if (!_xlrelarr)
+		return;
+
+	for (i = 1; i < _xlast; i++)
+	{
+		Relation	reln = &(_xlrelarr[i].reldata);
+		if (reln->rd_fd >= 0)
+			smgrclose(DEFAULT_SMGR, reln);
+	}
+
+	free(_xlrelarr);
+	free(_xlpgcarr);
+
+	hash_destroy(_xlrelcache);
+	_xlrelarr = NULL;
+}
+
+static void
+_xl_init_rel_cache(void)
+{
+	HASHCTL	ctl;
+
+	_xlcnt = _XLOG_INITRELCACHESIZE;
+	_xlast = 0;
+	_xlrelarr = (XLogRelDesc*) malloc(sizeof(XLogRelDesc) * _xlcnt);
+	memset(_xlrelarr, 0, sizeof(XLogRelDesc) * _xlcnt);
+	_xlpgcarr = (Form_pg_class) malloc(sizeof(FormData_pg_class) * _xlcnt);
+	memset(_xlpgcarr, 0, sizeof(XLogRelDesc) * _xlcnt);
+
+	memset(&ctl, 0, (int) sizeof(ctl));
+	ctl.keysize = sizeof(RelFileNode);
+	ctl.datasize = sizeof(XLogRelDesc*);
+	ctl.hash = tag_hash;
+
+	_xlrelcache = hash_create(_XLOG_INITRELCACHESIZE, &ctl,
+								HASH_ELEM | HASH_FUNCTION);
+}
+
+static XLogRelDesc*
+_xl_new_reldesc(void)
+{
+	_xlast++;
+	if (_xlast < _xlcnt)
+	{
+		_xlrelarr[_xlast].reldata.rd_rel = &(_xlpgcarr[_xlast]);
+		return(&(_xlrelarr[_xlast]));
+	}
+
+	if ( 2 * _xlcnt <= _XLOG_MAXRELCACHESIZE)
+	{
+		_xlrelarr = (XLogRelDesc*) realloc(_xlrelarr, 
+						2 * sizeof(XLogRelDesc) * _xlcnt);
+		memset(&(_xlrelarr[_xlcnt]), 0, sizeof(XLogRelDesc) * _xlcnt);
+		_xlpgcarr = (Form_pg_class) realloc(_xlpgcarr, 
+						2 * sizeof(FormData_pg_class) * _xlcnt);
+		memset(&(_xlpgcarr[_xlcnt]), 0, sizeof(FormData_pg_class) * _xlcnt);
+		_xlcnt += _xlcnt;
+		_xlrelarr[_xlast].reldata.rd_rel = &(_xlpgcarr[_xlast]);
+		return(&(_xlrelarr[_xlast]));
+	}
+	else /* reuse */
+	{
+		XLogRelCacheEntry	   *hentry;
+		bool					found;
+		XLogRelDesc			   *res = _xlrelarr[0].moreRecently;
+		Form_pg_class			tpgc = res->reldata.rd_rel;
+
+		res->lessRecently->moreRecently = res->moreRecently;
+		res->moreRecently->lessRecently = res->lessRecently;
+
+		hentry = (XLogRelCacheEntry*) hash_search(_xlrelcache, 
+			(char*)&(res->reldata.rd_node), HASH_REMOVE, &found);
+
+		if (hentry == NULL)
+			elog(STOP, "XLogOpenRelation: can't delete from cache");
+		if (!found)
+			elog(STOP, "XLogOpenRelation: file was not found in cache");
+
+		if (res->reldata.rd_fd >= 0)
+			smgrclose(DEFAULT_SMGR, &(res->reldata));
+
+		memset(res, 0, sizeof(XLogRelDesc));
+		memset(tpgc, 0, sizeof(FormData_pg_class));
+		res->reldata.rd_rel = tpgc;
+
+		_xlast--;
+		return(res);
+	}
+}
+
+Relation
+XLogOpenRelation(bool redo, RmgrId rmid, RelFileNode rnode)
+{
+	XLogRelDesc			   *res;
+	XLogRelCacheEntry	   *hentry;
+	bool					found;
+
+	if (!_xlrelarr)
+		_xl_init_rel_cache();
+
+	hentry = (XLogRelCacheEntry*) 
+			hash_search(_xlrelcache, (char*)&rnode, HASH_FIND, &found);
+
+	if (hentry == NULL)
+		elog(STOP, "XLogOpenRelation: error in cache");
+
+	if (found)
+	{
+		res = hentry->rdesc;
+
+		res->lessRecently->moreRecently = res->moreRecently;
+		res->moreRecently->lessRecently = res->lessRecently;
+	}
+	else
+	{
+		res = _xl_new_reldesc();
+
+		sprintf(RelationGetPhysicalRelationName(&(res->reldata)), "%u", rnode.relNode);
+
+		/* unexisting DB id */
+		res->reldata.rd_lockInfo.lockRelId.dbId = RecoveryDb;
+		res->reldata.rd_lockInfo.lockRelId.relId = rnode.relNode;
+		res->reldata.rd_node = rnode;
+
+		hentry = (XLogRelCacheEntry*) 
+			hash_search(_xlrelcache, (char*)&rnode, HASH_ENTER, &found);
+
+		if (hentry == NULL)
+			elog(STOP, "XLogOpenRelation: can't insert into cache");
+
+		if (found)
+			elog(STOP, "XLogOpenRelation: file found on insert into cache");
+
+		hentry->rdesc = res;
+
+		res->reldata.rd_unlinked = true;	/* look smgropen */
+		res->reldata.rd_fd = smgropen(DEFAULT_SMGR, &(res->reldata));
+	}
+
+	res->moreRecently = &(_xlrelarr[0]);
+	res->lessRecently = _xlrelarr[0].lessRecently;
+	_xlrelarr[0].lessRecently = res;
+	res->lessRecently->moreRecently = res;
+
+	if (res->reldata.rd_fd < 0)		/* file doesn't exist */
+		return(NULL);
+
+	return(&(res->reldata));
+}
+
 #endif