unl2log_20110422.patch

diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
new file mode 100644
index 2812681..d285cf2
*** a/src/backend/access/transam/twophase.c
--- b/src/backend/access/transam/twophase.c
*************** static void RecordTransactionCommitPrepa
*** 151,156 ****
--- 151,158 ----
  								RelFileNode *rels,
  								int ninvalmsgs,
  								SharedInvalidationMessage *invalmsgs,
+ 								int ninitforks,
+ 								RelFileNode *initforks,
  								bool initfileinval);
  static void RecordTransactionAbortPrepared(TransactionId xid,
  							   int nchildren,
*************** typedef struct TwoPhaseFileHeader
*** 769,774 ****
--- 771,777 ----
  	int32		ncommitrels;	/* number of delete-on-commit rels */
  	int32		nabortrels;		/* number of delete-on-abort rels */
  	int32		ninvalmsgs;		/* number of cache invalidation messages */
+ 	int32		ncommitinitforks;/* number of delete-on-commit init fork rels */
  	bool		initfileinval;	/* does relcache init file need invalidation? */
  	char		gid[GIDSIZE];	/* GID for transaction */
  } TwoPhaseFileHeader;
*************** StartPrepare(GlobalTransaction gxact)
*** 845,850 ****
--- 848,854 ----
  	TransactionId *children;
  	RelFileNode *commitrels;
  	RelFileNode *abortrels;
+ 	RelFileNode *commitinitforks;
  	SharedInvalidationMessage *invalmsgs;
  
  	/* Initialize linked list */
*************** StartPrepare(GlobalTransaction gxact)
*** 872,877 ****
--- 876,882 ----
  	hdr.nabortrels = smgrGetPendingDeletes(false, &abortrels);
  	hdr.ninvalmsgs = xactGetCommittedInvalidationMessages(&invalmsgs,
  														  &hdr.initfileinval);
+ 	hdr.ncommitinitforks = smgrGetPendingInitForkDeletes(true, &commitinitforks);
  	StrNCpy(hdr.gid, gxact->gid, GIDSIZE);
  
  	save_state_data(&hdr, sizeof(TwoPhaseFileHeader));
*************** StartPrepare(GlobalTransaction gxact)
*** 902,907 ****
--- 907,918 ----
  						hdr.ninvalmsgs * sizeof(SharedInvalidationMessage));
  		pfree(invalmsgs);
  	}
+ 	if (hdr.ncommitinitforks > 0)
+ 	{
+ 		save_state_data(commitinitforks,
+ 						hdr.ncommitinitforks * sizeof(RelFileNode));
+ 		pfree(commitinitforks);
+ 	}
  }
  
  /*
*************** FinishPreparedTransaction(const char *gi
*** 1250,1255 ****
--- 1261,1267 ----
  	RelFileNode *commitrels;
  	RelFileNode *abortrels;
  	RelFileNode *delrels;
+ 	RelFileNode *commitinitforks;
  	int			ndelrels;
  	SharedInvalidationMessage *invalmsgs;
  	int			i;
*************** FinishPreparedTransaction(const char *gi
*** 1285,1290 ****
--- 1297,1304 ----
  	bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
  	invalmsgs = (SharedInvalidationMessage *) bufptr;
  	bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
+ 	commitinitforks = (RelFileNode *) bufptr;
+ 	bufptr += MAXALIGN(hdr->ncommitinitforks * sizeof(RelFileNode));
  
  	/* compute latestXid among all children */
  	latestXid = TransactionIdLatest(xid, hdr->nsubxacts, children);
*************** FinishPreparedTransaction(const char *gi
*** 1302,1307 ****
--- 1316,1323 ----
  										hdr->nsubxacts, children,
  										hdr->ncommitrels, commitrels,
  										hdr->ninvalmsgs, invalmsgs,
+ 										hdr->ncommitinitforks,
+ 										commitinitforks,
  										hdr->initfileinval);
  	else
  		RecordTransactionAbortPrepared(xid,
*************** FinishPreparedTransaction(const char *gi
*** 1349,1354 ****
--- 1365,1381 ----
  		smgrclose(srel);
  	}
  
+ 	if (isCommit)
+ 	{
+ 		/* Make sure init fork files supposed to be dropped are dropped */
+ 		for (i = 0; i < hdr->ncommitinitforks; i++)
+ 		{
+ 			SMgrRelation srel = smgropen(commitinitforks[i], InvalidBackendId);
+ 			if (smgrexists(srel, INIT_FORKNUM))
+ 				smgrdounlink(srel, INIT_FORKNUM, true);
+ 			smgrclose(srel);
+ 		}
+ 	}
  	/*
  	 * Handle cache invalidation messages.
  	 *
*************** RecoverPreparedTransactions(void)
*** 1890,1895 ****
--- 1917,1923 ----
  			bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
  			bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
  			bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
+ 			bufptr += MAXALIGN(hdr->ncommitinitforks * sizeof(RelFileNode));
  
  			/*
  			 * It's possible that SubTransSetParent has been set before, if
*************** RecordTransactionCommitPrepared(Transact
*** 1961,1969 ****
  								RelFileNode *rels,
  								int ninvalmsgs,
  								SharedInvalidationMessage *invalmsgs,
  								bool initfileinval)
  {
! 	XLogRecData rdata[4];
  	int			lastrdata = 0;
  	xl_xact_commit_prepared xlrec;
  	XLogRecPtr	recptr;
--- 1989,1999 ----
  								RelFileNode *rels,
  								int ninvalmsgs,
  								SharedInvalidationMessage *invalmsgs,
+ 								int ninitforks,
+ 								RelFileNode *initforks,
  								bool initfileinval)
  {
! 	XLogRecData rdata[5];
  	int			lastrdata = 0;
  	xl_xact_commit_prepared xlrec;
  	XLogRecPtr	recptr;
*************** RecordTransactionCommitPrepared(Transact
*** 1981,1986 ****
--- 2011,2017 ----
  	xlrec.crec.nrels = nrels;
  	xlrec.crec.nsubxacts = nchildren;
  	xlrec.crec.nmsgs = ninvalmsgs;
+ 	xlrec.crec.ninitforks = ninitforks;
  
  	rdata[0].data = (char *) (&xlrec);
  	rdata[0].len = MinSizeOfXactCommitPrepared;
*************** RecordTransactionCommitPrepared(Transact
*** 2012,2017 ****
--- 2043,2057 ----
  		rdata[3].buffer = InvalidBuffer;
  		lastrdata = 3;
  	}
+ 	/* dump init fork deletions */
+ 	if (ninitforks > 0)
+ 	{
+ 		rdata[lastrdata].next = &(rdata[4]);
+ 		rdata[4].data = (char *) initforks;
+ 		rdata[4].len = ninitforks * sizeof(RelFileNode);
+ 		rdata[4].buffer = InvalidBuffer;
+ 		lastrdata = 4;
+ 	}
  	rdata[lastrdata].next = NULL;
  
  	recptr = XLogInsert(RM_XACT_ID, XLOG_XACT_COMMIT_PREPARED, rdata);
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
new file mode 100644
index 8a4c4ec..a0bb222
*** a/src/backend/access/transam/xact.c
--- b/src/backend/access/transam/xact.c
*************** RecordTransactionCommit(void)
*** 907,912 ****
--- 907,914 ----
  	TransactionId latestXid = InvalidTransactionId;
  	int			nrels;
  	RelFileNode *rels;
+ 	int			ninitforks;
+ 	RelFileNode *initforks;
  	int			nchildren;
  	TransactionId *children;
  	int			nmsgs = 0;
*************** RecordTransactionCommit(void)
*** 917,922 ****
--- 919,926 ----
  	/* Get data needed for commit record */
  	nrels = smgrGetPendingDeletes(true, &rels);
  	nchildren = xactGetCommittedChildren(&children);
+ 	/* TODO 2PC needs this too? */
+ 	ninitforks = smgrGetPendingInitForkDeletes(true, &initforks);
  	if (XLogStandbyInfoActive())
  		nmsgs = xactGetCommittedInvalidationMessages(&invalMessages,
  													 &RelcacheInitFileInval);
*************** RecordTransactionCommit(void)
*** 955,961 ****
  		/*
  		 * Begin commit critical section and insert the commit XLOG record.
  		 */
! 		XLogRecData rdata[4];
  		int			lastrdata = 0;
  		xl_xact_commit xlrec;
  
--- 959,965 ----
  		/*
  		 * Begin commit critical section and insert the commit XLOG record.
  		 */
! 		XLogRecData rdata[5];
  		int			lastrdata = 0;
  		xl_xact_commit xlrec;
  
*************** RecordTransactionCommit(void)
*** 999,1004 ****
--- 1003,1009 ----
  		xlrec.nrels = nrels;
  		xlrec.nsubxacts = nchildren;
  		xlrec.nmsgs = nmsgs;
+ 		xlrec.ninitforks = ninitforks;
  		rdata[0].data = (char *) (&xlrec);
  		rdata[0].len = MinSizeOfXactCommit;
  		rdata[0].buffer = InvalidBuffer;
*************** RecordTransactionCommit(void)
*** 1029,1034 ****
--- 1034,1048 ----
  			rdata[3].buffer = InvalidBuffer;
  			lastrdata = 3;
  		}
+ 		/* dump init fork deletions */
+ 		if (ninitforks > 0)
+ 		{
+ 			rdata[lastrdata].next = &(rdata[4]);
+ 			rdata[4].data = (char *) initforks;
+ 			rdata[4].len = ninitforks * sizeof(RelFileNode);
+ 			rdata[4].buffer = InvalidBuffer;
+ 			lastrdata = 4;
+ 		}
  		rdata[lastrdata].next = NULL;
  
  		(void) XLogInsert(RM_XACT_ID, XLOG_XACT_COMMIT, rdata);
*************** cleanup:
*** 1143,1148 ****
--- 1157,1164 ----
  	/* Clean up local data */
  	if (rels)
  		pfree(rels);
+ 	if (initforks)
+ 		pfree(initforks);
  
  	return latestXid;
  }
*************** xact_redo_commit(xl_xact_commit *xlrec, 
*** 4438,4443 ****
--- 4454,4460 ----
  {
  	TransactionId *sub_xids;
  	SharedInvalidationMessage *inval_msgs;
+ 	RelFileNode *initforks;
  	TransactionId max_xid;
  	int			i;
  
*************** xact_redo_commit(xl_xact_commit *xlrec, 
*** 4445,4450 ****
--- 4462,4469 ----
  	sub_xids = (TransactionId *) &(xlrec->xnodes[xlrec->nrels]);
  	/* invalidation messages array follows subxids */
  	inval_msgs = (SharedInvalidationMessage *) &(sub_xids[xlrec->nsubxacts]);
+ 	/* init fork deletes array follows invalidation messages */
+ 	initforks = (RelFileNode *) &(inval_msgs[xlrec->nmsgs]);
  
  	max_xid = TransactionIdLatest(xid, xlrec->nsubxacts, sub_xids);
  
*************** xact_redo_commit(xl_xact_commit *xlrec, 
*** 4534,4539 ****
--- 4553,4570 ----
  		smgrclose(srel);
  	}
  
+ 	/* Make sure init fork files supposed to be dropped are dropped */
+ 	for (i = 0; i < xlrec->ninitforks; i++)
+ 	{
+ 		SMgrRelation srel = smgropen(initforks[i], InvalidBackendId);
+ 		if (smgrexists(srel, INIT_FORKNUM))
+ 		{
+ 			XLogDropRelation(initforks[i], INIT_FORKNUM);
+ 			smgrdounlink(srel, INIT_FORKNUM, true);
+ 		}
+ 		smgrclose(srel);
+ 	}
+ 
  	/*
  	 * We issue an XLogFlush() for the same reason we emit ForceSyncCommit()
  	 * in normal operation. For example, in DROP DATABASE, we delete all the
*************** xact_desc_commit(StringInfo buf, xl_xact
*** 4749,4754 ****
--- 4780,4801 ----
  				appendStringInfo(buf, " unknown id %d", msg->id);
  		}
  	}
+ 	if (xlrec->ninitforks > 0)
+ 	{
+ 		RelFileNode *rl;
+ 
+ 		rl = (RelFileNode *) &xacts[xlrec->nmsgs];
+ 
+ 		appendStringInfo(buf, "; init fork rels:");
+ 		for (i = 0; i < xlrec->ninitforks; i++)
+ 		{
+ 			char	   *path = relpathperm(rl[i], INIT_FORKNUM);
+ 
+ 			appendStringInfo(buf, " %s", path);
+ 			pfree(path);
+ 		}
+ 	}
+ 
  }
  
  static void
diff --git a/src/backend/catalog/storage.c b/src/backend/catalog/storage.c
new file mode 100644
index 57987be..3363d7e
*** a/src/backend/catalog/storage.c
--- b/src/backend/catalog/storage.c
*************** typedef struct PendingRelDelete
*** 55,60 ****
--- 55,61 ----
  	BackendId	backend;		/* InvalidBackendId if not a temp rel */
  	bool		atCommit;		/* T=delete at commit; F=delete at abort */
  	int			nestLevel;		/* xact nesting level of request */
+ 	bool		onlyInitFork;	/* delete only INIT_FORKNUM */
  	struct PendingRelDelete *next;		/* linked-list link */
  } PendingRelDelete;
  
*************** RelationCreateStorage(RelFileNode rnode,
*** 135,140 ****
--- 136,142 ----
  	pending->backend = backend;
  	pending->atCommit = false;	/* delete if abort */
  	pending->nestLevel = GetCurrentTransactionNestLevel();
+ 	pending->onlyInitFork = false;
  	pending->next = pendingDeletes;
  	pendingDeletes = pending;
  }
*************** log_smgrcreate(RelFileNode *rnode, ForkN
*** 163,173 ****
  }
  
  /*
!  * RelationDropStorage
   *		Schedule unlinking of physical storage at transaction commit.
   */
! void
! RelationDropStorage(Relation rel)
  {
  	PendingRelDelete *pending;
  
--- 165,176 ----
  }
  
  /*
!  * RelationDropStorageForks
   *		Schedule unlinking of physical storage at transaction commit.
+  *	onlyInitFork: true if only the init fork as to be deleted
   */
! static void
! RelationDropStorageForks(Relation rel, bool onlyInitFork)
  {
  	PendingRelDelete *pending;
  
*************** RelationDropStorage(Relation rel)
*** 178,183 ****
--- 181,187 ----
  	pending->backend = rel->rd_backend;
  	pending->atCommit = true;	/* delete if commit */
  	pending->nestLevel = GetCurrentTransactionNestLevel();
+ 	pending->onlyInitFork = onlyInitFork;
  	pending->next = pendingDeletes;
  	pendingDeletes = pending;
  
*************** RelationDropStorage(Relation rel)
*** 195,200 ****
--- 199,225 ----
  }
  
  /*
+  * RelationDropStorage
+  *		Schedule unlinking of physical storage at transaction commit.
+  */
+ void
+ RelationDropStorage(Relation rel)
+ {
+ 	RelationDropStorageForks(rel, false);
+ }
+ 
+ /*
+  * RelationDropInitForkStorage
+  *		Schedule unlinking of physical storage of the init fork at transaction
+  *		commit.
+  */
+ void
+ RelationDropInitForkStorage(Relation rel)
+ {
+ 	RelationDropStorageForks(rel, true);
+ }
+ 
+ /*
   * RelationPreserveStorage
   *		Mark a relation as not to be deleted after all.
   *
*************** smgrDoPendingDeletes(bool isCommit)
*** 358,367 ****
  				int			i;
  
  				srel = smgropen(pending->relnode, pending->backend);
! 				for (i = 0; i <= MAX_FORKNUM; i++)
  				{
! 					if (smgrexists(srel, i))
! 						smgrdounlink(srel, i, false);
  				}
  				smgrclose(srel);
  			}
--- 383,400 ----
  				int			i;
  
  				srel = smgropen(pending->relnode, pending->backend);
! 				if (pending->onlyInitFork)
  				{
! 					if (smgrexists(srel, INIT_FORKNUM))
! 						smgrdounlink(srel, INIT_FORKNUM, false);
! 				}
! 				else
! 				{
! 					for (i = 0; i <= MAX_FORKNUM; i++)
! 					{
! 						if (smgrexists(srel, i))
! 							smgrdounlink(srel, i, false);
! 					}
  				}
  				smgrclose(srel);
  			}
*************** smgrDoPendingDeletes(bool isCommit)
*** 373,378 ****
--- 406,449 ----
  }
  
  /*
+  * smgrGetPendingInitForkDeletes() -- Get a list of init forksto be deleted.
+ */
+ int
+ smgrGetPendingInitForkDeletes(bool forCommit, RelFileNode **ptr)
+ {
+ 	int			nestLevel = GetCurrentTransactionNestLevel();
+ 	int			nrels;
+ 	RelFileNode *rptr;
+ 	PendingRelDelete *pending;
+ 
+ 	nrels = 0;
+ 	for (pending = pendingDeletes; pending != NULL; pending = pending->next)
+ 	{
+ 		if (pending->nestLevel >= nestLevel && pending->atCommit == forCommit
+ 			&& pending->backend == InvalidBackendId && pending->onlyInitFork)
+ 			nrels++;
+ 	}
+ 	if (nrels == 0)
+ 	{
+ 		*ptr = NULL;
+ 		return 0;
+ 	}
+ 	rptr = (RelFileNode *) palloc(nrels * sizeof(RelFileNode));
+ 	*ptr = rptr;
+ 	for (pending = pendingDeletes; pending != NULL; pending = pending->next)
+ 	{
+ 		if (pending->nestLevel >= nestLevel && pending->atCommit == forCommit
+ 			&& pending->backend == InvalidBackendId && pending->onlyInitFork)
+ 		{
+ 			*rptr = pending->relnode;
+ 			rptr++;
+ 		}
+ 	}
+ 	return nrels;
+ }
+ 
+ 
+ /*
   * smgrGetPendingDeletes() -- Get a list of non-temp relations to be deleted.
   *
   * The return value is the number of relations scheduled for termination.
*************** smgrGetPendingDeletes(bool forCommit, Re
*** 401,407 ****
  	for (pending = pendingDeletes; pending != NULL; pending = pending->next)
  	{
  		if (pending->nestLevel >= nestLevel && pending->atCommit == forCommit
! 			&& pending->backend == InvalidBackendId)
  			nrels++;
  	}
  	if (nrels == 0)
--- 472,478 ----
  	for (pending = pendingDeletes; pending != NULL; pending = pending->next)
  	{
  		if (pending->nestLevel >= nestLevel && pending->atCommit == forCommit
! 			&& pending->backend == InvalidBackendId && !pending->onlyInitFork)
  			nrels++;
  	}
  	if (nrels == 0)
*************** smgrGetPendingDeletes(bool forCommit, Re
*** 414,420 ****
  	for (pending = pendingDeletes; pending != NULL; pending = pending->next)
  	{
  		if (pending->nestLevel >= nestLevel && pending->atCommit == forCommit
! 			&& pending->backend == InvalidBackendId)
  		{
  			*rptr = pending->relnode;
  			rptr++;
--- 485,491 ----
  	for (pending = pendingDeletes; pending != NULL; pending = pending->next)
  	{
  		if (pending->nestLevel >= nestLevel && pending->atCommit == forCommit
! 			&& pending->backend == InvalidBackendId && !pending->onlyInitFork)
  		{
  			*rptr = pending->relnode;
  			rptr++;
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
new file mode 100644
index 790bc2a..bbdcc64
*** a/src/backend/commands/tablecmds.c
--- b/src/backend/commands/tablecmds.c
*************** static void ATPrepAddInherit(Relation ch
*** 358,363 ****
--- 358,365 ----
  static void ATExecAddInherit(Relation child_rel, RangeVar *parent, LOCKMODE lockmode);
  static void ATExecDropInherit(Relation rel, RangeVar *parent, LOCKMODE lockmode);
  static void ATExecGenericOptions(Relation rel, List *options);
+ static void ATPrepSetLogged(Relation rel);
+ static void ATExecSetLogged(Relation rel);
  
  static void copy_relation_data(SMgrRelation rel, SMgrRelation dst,
  				   ForkNumber forkNum, char relpersistence);
*************** AlterTableGetLockLevel(List *cmds)
*** 2614,2619 ****
--- 2616,2622 ----
  			case AT_DropNotNull:		/* may change some SQL plans */
  			case AT_SetNotNull:
  			case AT_GenericOptions:
+ 			case AT_SetLogged:				/* must drop shared buffers */
  				cmd_lockmode = AccessExclusiveLock;
  				break;
  
*************** ATPrepCmd(List **wqueue, Relation rel, A
*** 2958,2963 ****
--- 2961,2973 ----
  			/* No command-specific prep needed */
  			pass = AT_PASS_MISC;
  			break;
+ 		case AT_SetLogged:	/* ALTER TABLE SET LOGGED */
+ 			ATSimpleRecursion(wqueue, rel, cmd, recurse, lockmode);
+ 			ATSimplePermissions(rel, ATT_TABLE);
+ 			/* Performs also own permission checks */
+ 			ATPrepSetLogged(rel);
+ 			pass = AT_PASS_MISC;
+ 			break;
  		default:				/* oops */
  			elog(ERROR, "unrecognized alter table type: %d",
  				 (int) cmd->subtype);
*************** ATExecCmd(List **wqueue, AlteredTableInf
*** 3221,3226 ****
--- 3231,3239 ----
  		case AT_GenericOptions:
  			ATExecGenericOptions(rel, (List *) cmd->def);
  			break;
+ 		case AT_SetLogged:
+ 			ATExecSetLogged(rel);
+ 			break;
  		default:				/* oops */
  			elog(ERROR, "unrecognized alter table type: %d",
  				 (int) cmd->subtype);
*************** ATPrepSetTableSpace(AlteredTableInfo *ta
*** 7603,7608 ****
--- 7616,7800 ----
  }
  
  /*
+  * 0) check we're in the wal_level=minimal case - this check will have to be
+  * removed later...
+  * 1) check the table is, in fact, unlogged
+  * 2) unlogged referencing permanent is OK, but permanent referencing unlogged
+  * is a no-no, so what matters when upgrading is "outbound" foreign keys
+  */
+ static void
+ ATPrepSetLogged(Relation rel)
+ {
+ 	Relation	pg_constraint;
+ 	HeapTuple	tuple;
+ 	SysScanDesc scan;
+ 	ScanKeyData skey[1];
+ 
+ 	if (wal_level != WAL_LEVEL_MINIMAL)
+ 	{
+ 		/* TODO change ERRCODE_INSUFFICIENT_PRIVILEGE to something else */
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ 				 errmsg("cannot change table to LOGGED in wal_level != minimal case")));
+ 
+ 	}
+ 
+ 	if (rel->rd_rel->relpersistence != RELPERSISTENCE_UNLOGGED)
+ 	{
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+ 				 errmsg("table %s is not unlogged",
+ 						 RelationGetRelationName(rel))));
+ 	}
+ 
+ 	/*
+ 	 * Fetch the constraint tuple from pg_constraint.
+ 	 */
+ 	pg_constraint = heap_open(ConstraintRelationId, AccessShareLock);
+ 
+ 	ScanKeyInit(&skey[0],
+ 				Anum_pg_constraint_conrelid,
+ 				BTEqualStrategyNumber, F_OIDEQ,
+ 				ObjectIdGetDatum(RelationGetRelid(rel)));
+ 
+ 	scan = systable_beginscan(pg_constraint, ConstraintRelidIndexId, true,
+ 							  SnapshotNow, 1, skey);
+ 
+ 	while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+ 	{
+ 		Form_pg_constraint con = (Form_pg_constraint) GETSTRUCT(tuple);
+ 		if (con->contype == CONSTRAINT_FOREIGN)
+ 		{
+ 			Relation relfk = relation_open(con->confrelid, AccessShareLock);
+ 			if (relfk->rd_rel->relpersistence != RELPERSISTENCE_PERMANENT)
+ 			{
+ 				/* TODO better error description */
+ 				/* TODO: exit at first error, or list all dependencies? */
+ 				ereport(ERROR,
+ 						(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+ 						 errmsg("table %s references unlogged table %s",
+ 								 RelationGetRelationName(rel),
+ 								 RelationGetRelationName(relfk))));
+ 			}
+ 			relation_close(relfk, AccessShareLock);
+ 		}
+ 	}
+ 
+ 	systable_endscan(scan);
+ 
+ 	heap_close(pg_constraint, AccessShareLock);
+ 
+ }
+ 
+ /*
+ * 1) Write out all shared buffers for the target table/index, and drop them
+ * 2) fsync() any segments of the target relation - of any fork except
+ * that init fork
+ * 3) Arrange for the appropriate file deletions at commit or abort, by
+ * updating pendingDeletes
+ */
+ static void
+ ATUpdateToLogged(Relation rel, Relation	pg_classRel)
+ {
+ 	HeapTuple	reltup;
+ 	Form_pg_class relStruct;
+ 	bool fsm;
+ 	bool vm;
+ 
+ 	/* main heap */
+ 	FlushRelationBuffers(rel);
+ 
+ 	/* FlushRelationBuffers will have opened rd_smgr */
+ 	DropRelFileNodeBuffers(rel->rd_smgr->smgr_rnode, MAIN_FORKNUM, 0);
+ 	smgrimmedsync(rel->rd_smgr, MAIN_FORKNUM);
+ 
+ 	fsm = smgrexists(rel->rd_smgr, FSM_FORKNUM);
+ 	if (fsm)
+ 		smgrimmedsync(rel->rd_smgr, FSM_FORKNUM);
+ 
+ 	vm = smgrexists(rel->rd_smgr, VISIBILITYMAP_FORKNUM);
+ 	if (vm)
+ 		smgrimmedsync(rel->rd_smgr, VISIBILITYMAP_FORKNUM);
+ 
+ 	/* init fork delete */
+ 	RelationDropInitForkStorage(rel);
+ 
+ 	reltup = SearchSysCacheCopy1(RELOID,
+ 								 ObjectIdGetDatum(RelationGetRelid(rel)));
+ 	if (!HeapTupleIsValid(reltup))
+ 		elog(ERROR, "cache lookup failed for relation %u",
+ 			 RelationGetRelid(rel));
+ 	relStruct = (Form_pg_class) GETSTRUCT(reltup);
+ 
+ 	Assert(relStruct->relpersistence == RELPERSISTENCE_UNLOGGED);
+ 
+ 	relStruct->relpersistence = RELPERSISTENCE_PERMANENT;
+ 
+ 	simple_heap_update(pg_classRel, &reltup->t_self, reltup);
+ 
+ 	/* keep catalog indexes current */
+ 	CatalogUpdateIndexes(pg_classRel, reltup);
+ 
+ 	heap_freetuple(reltup);
+ 
+ }
+ 
+ /*
+  * update main heap, tost and index(es) of the target Relation to logged
+ */
+ static void
+ ATExecSetLogged(Relation rel)
+ {
+ 	Relation	indexRelation;
+ 	ScanKeyData skey;
+ 	SysScanDesc scan;
+ 	HeapTuple	indexTuple;
+ 	Relation	relrel;
+ 
+ 	/* open pg_class to update relpersistence */
+ 	relrel = heap_open(RelationRelationId, RowExclusiveLock);
+ 
+ 	/* main heap */
+ 	ATUpdateToLogged(rel, relrel);
+ 
+ 	/* toast heap, if any */
+ 	if (OidIsValid(rel->rd_rel->reltoastrelid))
+ 	{
+ 		Relation	toastrel;
+ 
+ 		toastrel = heap_open(rel->rd_rel->reltoastrelid, AccessShareLock);
+ 		ATUpdateToLogged(toastrel, relrel);
+ 		heap_close(toastrel, AccessShareLock);
+ 	}
+ 
+ 	/* indexes init fork delete */
+ 	/* Prepare to scan pg_index for entries having indrelid = this rel. */
+ 	indexRelation = heap_open(IndexRelationId, AccessShareLock);
+ 	ScanKeyInit(&skey,
+ 				Anum_pg_index_indrelid,
+ 				BTEqualStrategyNumber, F_OIDEQ,
+ 				ObjectIdGetDatum(RelationGetRelid(rel)));
+ 
+ 	scan = systable_beginscan(indexRelation, IndexIndrelidIndexId, true,
+ 							  SnapshotNow, 1, &skey);
+ 
+ 	while (HeapTupleIsValid(indexTuple = systable_getnext(scan)))
+ 	{
+ 		Form_pg_index index = (Form_pg_index) GETSTRUCT(indexTuple);
+ 		Relation indxrel = index_open(index->indexrelid, AccessShareLock);
+ 
+ 		ATUpdateToLogged(indxrel, relrel);
+ 
+ 		index_close(indxrel, AccessShareLock);
+ 	}
+ 
+ 	systable_endscan(scan);
+ 	heap_close(indexRelation, AccessShareLock);
+ 
+ 	heap_close(relrel, RowExclusiveLock);
+ }
+ 
+ /*
   * ALTER TABLE/INDEX SET (...) or RESET (...)
   */
  static void
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
new file mode 100644
index a22ab66..e1a50a4
*** a/src/backend/parser/gram.y
--- b/src/backend/parser/gram.y
*************** static void SplitColQualList(List *qualL
*** 520,526 ****
  
  	LABEL LANGUAGE LARGE_P LAST_P LC_COLLATE_P LC_CTYPE_P LEADING
  	LEAST LEFT LEVEL LIKE LIMIT LISTEN LOAD LOCAL LOCALTIME LOCALTIMESTAMP
! 	LOCATION LOCK_P
  
  	MAPPING MATCH MAXVALUE MINUTE_P MINVALUE MODE MONTH_P MOVE
  
--- 520,526 ----
  
  	LABEL LANGUAGE LARGE_P LAST_P LC_COLLATE_P LC_CTYPE_P LEADING
  	LEAST LEFT LEVEL LIKE LIMIT LISTEN LOAD LOCAL LOCALTIME LOCALTIMESTAMP
! 	LOCATION LOCK_P LOGGED
  
  	MAPPING MATCH MAXVALUE MINUTE_P MINVALUE MODE MONTH_P MOVE
  
*************** alter_table_cmd:
*** 1795,1800 ****
--- 1795,1807 ----
  					n->missing_ok = FALSE;
  					$$ = (Node *)n;
  				}
+ 			/* ALTER TABLE <name> SET LOGGED  */
+ 			| SET LOGGED
+ 				{
+ 					AlterTableCmd *n = makeNode(AlterTableCmd);
+ 					n->subtype = AT_SetLogged;
+ 					$$ = (Node *)n;
+ 				}
  			/* ALTER TABLE <name> SET WITH OIDS  */
  			| SET WITH OIDS
  				{
*************** unreserved_keyword:
*** 11935,11940 ****
--- 11942,11948 ----
  			| LOCAL
  			| LOCATION
  			| LOCK_P
+ 			| LOGGED
  			| MAPPING
  			| MATCH
  			| MAXVALUE
diff --git a/src/include/access/xact.h b/src/include/access/xact.h
new file mode 100644
index cb440d4..ef56d17
*** a/src/include/access/xact.h
--- b/src/include/access/xact.h
*************** typedef struct xl_xact_commit
*** 123,134 ****
--- 123,136 ----
  	int			nrels;			/* number of RelFileNodes */
  	int			nsubxacts;		/* number of subtransaction XIDs */
  	int			nmsgs;			/* number of shared inval msgs */
+ 	int			ninitforks;  /* number of init fork RelFileNodes */
  	Oid			dbId;			/* MyDatabaseId */
  	Oid			tsId;			/* MyDatabaseTableSpace */
  	/* Array of RelFileNode(s) to drop at commit */
  	RelFileNode xnodes[1];		/* VARIABLE LENGTH ARRAY */
  	/* ARRAY OF COMMITTED SUBTRANSACTION XIDs FOLLOWS */
  	/* ARRAY OF SHARED INVALIDATION MESSAGES FOLLOWS */
+ 	/* ARRAY OF INIT FORK RelFileNodes FOLLOWS */
  } xl_xact_commit;
  
  #define MinSizeOfXactCommit offsetof(xl_xact_commit, xnodes)
diff --git a/src/include/catalog/storage.h b/src/include/catalog/storage.h
new file mode 100644
index 8dee8cf..04e1118
*** a/src/include/catalog/storage.h
--- b/src/include/catalog/storage.h
***************
*** 22,27 ****
--- 22,28 ----
  
  extern void RelationCreateStorage(RelFileNode rnode, char relpersistence);
  extern void RelationDropStorage(Relation rel);
+ extern void RelationDropInitForkStorage(Relation rel);
  extern void RelationPreserveStorage(RelFileNode rnode);
  extern void RelationTruncate(Relation rel, BlockNumber nblocks);
  
*************** extern void RelationTruncate(Relation re
*** 31,36 ****
--- 32,38 ----
   */
  extern void smgrDoPendingDeletes(bool isCommit);
  extern int	smgrGetPendingDeletes(bool forCommit, RelFileNode **ptr);
+ extern int	smgrGetPendingInitForkDeletes(bool forCommit, RelFileNode **ptr);
  extern void AtSubCommit_smgr(void);
  extern void AtSubAbort_smgr(void);
  extern void PostPrepare_smgr(void);
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
new file mode 100644
index 566cd3a..56f1384
*** a/src/include/nodes/parsenodes.h
--- b/src/include/nodes/parsenodes.h
*************** typedef enum AlterTableType
*** 1219,1224 ****
--- 1219,1225 ----
  	AT_AddInherit,				/* INHERIT parent */
  	AT_DropInherit,				/* NO INHERIT parent */
  	AT_GenericOptions,			/* OPTIONS (...) */
+ 	AT_SetLogged,				/* SET LOGGED */
  } AlterTableType;
  
  typedef struct AlterTableCmd	/* one subcommand of an ALTER TABLE */
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
new file mode 100644
index 12c2faf..45e17ea
*** a/src/include/parser/kwlist.h
--- b/src/include/parser/kwlist.h
*************** PG_KEYWORD("localtime", LOCALTIME, RESER
*** 227,232 ****
--- 227,233 ----
  PG_KEYWORD("localtimestamp", LOCALTIMESTAMP, RESERVED_KEYWORD)
  PG_KEYWORD("location", LOCATION, UNRESERVED_KEYWORD)
  PG_KEYWORD("lock", LOCK_P, UNRESERVED_KEYWORD)
+ PG_KEYWORD("logged", LOGGED, UNRESERVED_KEYWORD)
  PG_KEYWORD("mapping", MAPPING, UNRESERVED_KEYWORD)
  PG_KEYWORD("match", MATCH, UNRESERVED_KEYWORD)
  PG_KEYWORD("maxvalue", MAXVALUE, UNRESERVED_KEYWORD)