ssi-ddl-1.patch

*** a/src/backend/catalog/index.c
--- b/src/backend/catalog/index.c
***************
*** 54,59 ****
--- 54,60 ----
  #include "parser/parser.h"
  #include "storage/bufmgr.h"
  #include "storage/lmgr.h"
+ #include "storage/predicate.h"
  #include "storage/procarray.h"
  #include "storage/smgr.h"
  #include "utils/builtins.h"
***************
*** 114,120 **** static void validate_index_heapscan(Relation heapRelation,
  						IndexInfo *indexInfo,
  						Snapshot snapshot,
  						v_i_state *state);
- static Oid	IndexGetRelation(Oid indexId);
  static void SetReindexProcessing(Oid heapOid, Oid indexOid);
  static void ResetReindexProcessing(void);
  static void SetReindexPending(List *indexes);
--- 115,120 ----
***************
*** 2721,2727 **** validate_index_heapscan(Relation heapRelation,
   * IndexGetRelation: given an index's relation OID, get the OID of the
   * relation it is an index on.	Uses the system cache.
   */
! static Oid
  IndexGetRelation(Oid indexId)
  {
  	HeapTuple	tuple;
--- 2721,2727 ----
   * IndexGetRelation: given an index's relation OID, get the OID of the
   * relation it is an index on.	Uses the system cache.
   */
! Oid
  IndexGetRelation(Oid indexId)
  {
  	HeapTuple	tuple;
***************
*** 2782,2787 **** reindex_index(Oid indexId, bool skip_constraint_checks)
--- 2782,2793 ----
  	 */
  	CheckTableNotInUse(iRel, "REINDEX INDEX");
  
+ 	/*
+ 	 * All predicate locks on the index are about to be made invalid.
+ 	 * Promote them to relation locks on the heap.
+ 	 */
+ 	TransferPredicateLocksToHeapRelation(iRel);
+ 
  	PG_TRY();
  	{
  		/* Suppress use of the target index while rebuilding it */
*** a/src/backend/storage/lmgr/predicate.c
--- b/src/backend/storage/lmgr/predicate.c
***************
*** 185,190 ****
--- 185,191 ----
  #include "access/twophase.h"
  #include "access/twophase_rmgr.h"
  #include "access/xact.h"
+ #include "catalog/index.h"
  #include "miscadmin.h"
  #include "storage/bufmgr.h"
  #include "storage/predicate.h"
***************
*** 975,982 **** InitPredicateLocks(void)
  	bool		found;
  
  	/*
! 	 * Compute size of predicate lock target hashtable.
! 	 * Note these calculations must agree with PredicateLockShmemSize!
  	 */
  	max_table_size = NPREDICATELOCKTARGETENTS();
  
--- 976,983 ----
  	bool		found;
  
  	/*
! 	 * Compute size of predicate lock target hashtable. Note these
! 	 * calculations must agree with PredicateLockShmemSize!
  	 */
  	max_table_size = NPREDICATELOCKTARGETENTS();
  
***************
*** 1028,1035 **** InitPredicateLocks(void)
  									  hash_flags);
  
  	/*
! 	 * Compute size for serializable transaction hashtable.
! 	 * Note these calculations must agree with PredicateLockShmemSize!
  	 */
  	max_table_size = (MaxBackends + max_prepared_xacts);
  
--- 1029,1036 ----
  									  hash_flags);
  
  	/*
! 	 * Compute size for serializable transaction hashtable. Note these
! 	 * calculations must agree with PredicateLockShmemSize!
  	 */
  	max_table_size = (MaxBackends + max_prepared_xacts);
  
***************
*** 2276,2288 **** PredicateLockTupleRowVersionLink(const Relation relation,
  				newxmin;
  
  	/*
! 	 * Bail out quickly if there are no serializable transactions
! 	 * running.
  	 *
! 	 * It's safe to do this check without taking any additional
! 	 * locks. Even if a serializable transaction starts concurrently,
! 	 * we know it can't take any SIREAD locks on the modified tuple
! 	 * because the caller is holding the associated buffer page lock.
  	 */
  	if (!TransactionIdIsValid(PredXact->SxactGlobalXmin))
  		return;
--- 2277,2288 ----
  				newxmin;
  
  	/*
! 	 * Bail out quickly if there are no serializable transactions running.
  	 *
! 	 * It's safe to do this check without taking any additional locks. Even if
! 	 * a serializable transaction starts concurrently, we know it can't take
! 	 * any SIREAD locks on the modified tuple because the caller is holding
! 	 * the associated buffer page lock.
  	 */
  	if (!TransactionIdIsValid(PredXact->SxactGlobalXmin))
  		return;
***************
*** 2622,2627 **** exit:
--- 2622,2839 ----
  	return !outOfShmem;
  }
  
+ /*
+  * For all connections, transfer all predicate locks for the given relation
+  * to a single relation lock on the heap.  For a heap relation that includes
+  * all locks on indexes; for an index it is just the locks on that one
+  * index.
+  *
+  * This requires grabbing a lot of LW locks and scanning the entire lock
+  * target table for matches.  That makes this more expensive than most
+  * functions, but it will only be called for DDL type commands and there are
+  * fast returns when no serializable transactions are active or the relation
+  * is temporary.
+  *
+  * We are not using the existing TransferPredicateLocksToNewTarget because
+  * it acquires its own locks on the partitions of the two targets invovled,
+  * and we'll already be holding all partition locks.
+  *
+  * We can't throw an error from here, because the call could be from a
+  * transaction which is not serializable.
+  */
+ void
+ TransferPredicateLocksToHeapRelation(const Relation relation)
+ {
+ 	HASH_SEQ_STATUS seqstat;
+ 	PREDICATELOCKTARGET *oldtarget;
+ 	PREDICATELOCKTARGET *heaptarget;
+ 	PREDICATELOCKTARGETTAG heaptargettag;
+ 	PREDICATELOCKTAG newpredlocktag;
+ 	Oid			dbId;
+ 	Oid			indexId;
+ 	Oid			heapId;
+ 	int			i;
+ 	bool		isSingleIndex;
+ 	bool		found;
+ 	uint32		reservedtargettaghash;
+ 	uint32		heaptargettaghash;
+ 
+ 	/*
+ 	 * Bail out quickly if there are no serializable transactions running. As
+ 	 * with PredicateLockTupleRowVersionLink, it's safe to check this without
+ 	 * taking locks because the caller is holding the buffer page lock.
+ 	 */
+ 	if (!TransactionIdIsValid(PredXact->SxactGlobalXmin))
+ 		return;
+ 
+ 	if (SkipSplitTracking(relation))
+ 		return;
+ 
+ 	dbId = relation->rd_node.dbNode;
+ 	if (relation->rd_index == NULL)
+ 	{
+ 		isSingleIndex = false;
+ 		indexId = InvalidOid;	/* quiet compiler warning */
+ 		heapId = relation->rd_id;
+ 	}
+ 	else
+ 	{
+ 		isSingleIndex = true;
+ 		indexId = relation->rd_id;
+ 		heapId = relation->rd_index->indrelid;
+ 	}
+ 	Assert(heapId != InvalidOid);
+ 	SET_PREDICATELOCKTARGETTAG_RELATION(heaptargettag,
+ 										dbId,
+ 										heapId);
+ 	heaptargettaghash = PredicateLockTargetTagHashCode(&heaptargettag);
+ 	heaptarget = NULL;			/* Retrieve first time needed, then keep. */
+ 
+ 	LWLockAcquire(SerializablePredicateLockListLock, LW_EXCLUSIVE);
+ 	for (i = 0; i < NUM_PREDICATELOCK_PARTITIONS; i++)
+ 		LWLockAcquire(FirstPredicateLockMgrLock + i, LW_EXCLUSIVE);
+ 	LWLockAcquire(SerializableXactHashLock, LW_EXCLUSIVE);
+ 
+ 	/*
+ 	 * Remove the reserved entry to give us scratch space, so we know we'll be
+ 	 * able to create the new lock target.
+ 	 */
+ 	reservedtargettaghash = PredicateLockTargetTagHashCode(&ReservedTargetTag);
+ 	hash_search_with_hash_value(PredicateLockTargetHash,
+ 								&ReservedTargetTag,
+ 								reservedtargettaghash,
+ 								HASH_REMOVE, &found);
+ 	Assert(found);
+ 
+ 	/* Scan through PredicateLockHash and copy contents */
+ 	hash_seq_init(&seqstat, PredicateLockTargetHash);
+ 
+ 	while ((oldtarget = (PREDICATELOCKTARGET *) hash_seq_search(&seqstat)))
+ 	{
+ 		PREDICATELOCK *oldpredlock;
+ 
+ 		/*
+ 		 * Check whether this is a target which needs attention.
+ 		 */
+ 		if (GET_PREDICATELOCKTARGETTAG_DB(oldtarget->tag) != dbId)
+ 			continue;			/* wrong database */
+ 		if (isSingleIndex)
+ 		{
+ 			if (GET_PREDICATELOCKTARGETTAG_RELATION(oldtarget->tag) != indexId)
+ 				continue;		/* not the index we're looking for */
+ 		}
+ 		else
+ 		{
+ 			if (GET_PREDICATELOCKTARGETTAG_RELATION(oldtarget->tag) == heapId)
+ 			{
+ 				if (GET_PREDICATELOCKTARGETTAG_TYPE(oldtarget->tag) == PREDLOCKTAG_RELATION)
+ 					continue;	/* already the right lock */
+ 			}
+ 			else
+ 			{
+ 				/* This is an index.  Is it for the right heap? */
+ 				if (IndexGetRelation(GET_PREDICATELOCKTARGETTAG_RELATION(oldtarget->tag))
+ 					!= heapId)
+ 					continue;	/* index on wrong heap relation */
+ 			}
+ 		}
+ 
+ 		/*
+ 		 * If we made it here, we have work to do.	We make sure the heap
+ 		 * relation lock exists, then we walk the list of predicate locks for
+ 		 * the old target we found, moving all locks to the heap relation lock
+ 		 * -- unless they already hold that.
+ 		 */
+ 
+ 		/*
+ 		 * First make sure we have the heap relation target.  We only need to
+ 		 * do this once.
+ 		 */
+ 		if (heaptarget == NULL)
+ 		{
+ 			heaptarget = hash_search_with_hash_value(PredicateLockTargetHash,
+ 													 &heaptargettag,
+ 													 heaptargettaghash,
+ 													 HASH_ENTER, &found);
+ 			Assert(heaptarget != NULL);
+ 			if (!found)
+ 				SHMQueueInit(&heaptarget->predicateLocks);
+ 			newpredlocktag.myTarget = heaptarget;
+ 		}
+ 
+ 		/*
+ 		 * Loop through moving locks from this target to the relation target.
+ 		 */
+ 		oldpredlock = (PREDICATELOCK *)
+ 			SHMQueueNext(&(oldtarget->predicateLocks),
+ 						 &(oldtarget->predicateLocks),
+ 						 offsetof(PREDICATELOCK, targetLink));
+ 		while (oldpredlock)
+ 		{
+ 			PREDICATELOCK *nextpredlock;
+ 			PREDICATELOCK *newpredlock;
+ 			SerCommitSeqNo oldCommitSeqNo = oldpredlock->commitSeqNo;
+ 
+ 			nextpredlock = (PREDICATELOCK *)
+ 				SHMQueueNext(&(oldtarget->predicateLocks),
+ 							 &(oldpredlock->targetLink),
+ 							 offsetof(PREDICATELOCK, targetLink));
+ 			newpredlocktag.myXact = oldpredlock->tag.myXact;
+ 
+ 			SHMQueueDelete(&(oldpredlock->xactLink));
+ 			/* No need for retail delete from oldtarget list. */
+ 			hash_search(PredicateLockHash,
+ 						&oldpredlock->tag,
+ 						HASH_REMOVE, &found);
+ 			Assert(found);
+ 
+ 			newpredlock = (PREDICATELOCK *)
+ 				hash_search_with_hash_value
+ 				(PredicateLockHash,
+ 				 &newpredlocktag,
+ 				 PredicateLockHashCodeFromTargetHashCode(&newpredlocktag,
+ 														 heaptargettaghash),
+ 				 HASH_ENTER_NULL, &found);
+ 			Assert(newpredlock != NULL);
+ 			if (!found)
+ 			{
+ 				SHMQueueInsertBefore(&(heaptarget->predicateLocks),
+ 									 &(newpredlock->targetLink));
+ 				SHMQueueInsertBefore(&(newpredlocktag.myXact->predicateLocks),
+ 									 &(newpredlock->xactLink));
+ 				newpredlock->commitSeqNo = oldCommitSeqNo;
+ 			}
+ 			else
+ 			{
+ 				if (newpredlock->commitSeqNo < oldCommitSeqNo)
+ 					newpredlock->commitSeqNo = oldCommitSeqNo;
+ 			}
+ 
+ 			Assert(newpredlock->commitSeqNo != 0);
+ 			Assert((newpredlock->commitSeqNo == InvalidSerCommitSeqNo)
+ 				   || (newpredlock->tag.myXact == OldCommittedSxact));
+ 
+ 			oldpredlock = nextpredlock;
+ 		}
+ 
+ 		hash_search(PredicateLockTargetHash, &oldtarget->tag, HASH_REMOVE, &found);
+ 		Assert(found);
+ 	}
+ 
+ 	/* Put the reserved entry back */
+ 	hash_search_with_hash_value(PredicateLockTargetHash,
+ 								&ReservedTargetTag,
+ 								reservedtargettaghash,
+ 								HASH_ENTER, &found);
+ 	Assert(!found);
+ 
+ 	/* Release locks in reverse order */
+ 	LWLockRelease(SerializableXactHashLock);
+ 	for (i = NUM_PREDICATELOCK_PARTITIONS - 1; i >= 0; i--)
+ 		LWLockRelease(FirstPredicateLockMgrLock + i);
+ 	LWLockRelease(SerializablePredicateLockListLock);
+ }
+ 
  
  /*
   *		PredicateLockPageSplit
***************
*** 2646,2655 **** PredicateLockPageSplit(const Relation relation, const BlockNumber oldblkno,
  	bool		success;
  
  	/*
! 	 * Bail out quickly if there are no serializable transactions
! 	 * running. As with PredicateLockTupleRowVersionLink, it's safe to
! 	 * check this without taking locks because the caller is holding
! 	 * the buffer page lock.
  	 */
  	if (!TransactionIdIsValid(PredXact->SxactGlobalXmin))
  		return;
--- 2858,2866 ----
  	bool		success;
  
  	/*
! 	 * Bail out quickly if there are no serializable transactions running. As
! 	 * with PredicateLockTupleRowVersionLink, it's safe to check this without
! 	 * taking locks because the caller is holding the buffer page lock.
  	 */
  	if (!TransactionIdIsValid(PredXact->SxactGlobalXmin))
  		return;
*** a/src/include/catalog/index.h
--- b/src/include/catalog/index.h
***************
*** 66,71 **** extern void index_drop(Oid indexId);
--- 66,73 ----
  
  extern IndexInfo *BuildIndexInfo(Relation index);
  
+ extern Oid IndexGetRelation(Oid indexId);
+ 
  extern void FormIndexDatum(IndexInfo *indexInfo,
  			   TupleTableSlot *slot,
  			   EState *estate,
*** a/src/include/storage/predicate.h
--- b/src/include/storage/predicate.h
***************
*** 50,55 **** extern void PredicateLockTuple(const Relation relation, const HeapTuple tuple);
--- 50,56 ----
  extern void PredicateLockTupleRowVersionLink(const Relation relation, const HeapTuple oldTuple, const HeapTuple newTuple);
  extern void PredicateLockPageSplit(const Relation relation, const BlockNumber oldblkno, const BlockNumber newblkno);
  extern void PredicateLockPageCombine(const Relation relation, const BlockNumber oldblkno, const BlockNumber newblkno);
+ extern void TransferPredicateLocksToHeapRelation(const Relation relation);
  extern void ReleasePredicateLocks(const bool isCommit);
  
  /* conflict detection (may also trigger rollback) */