index-constraints-20090705.patch

diff --git a/src/backend/access/index/indexam.c b/src/backend/access/index/indexam.c
index 1515d9f..eedb456 100644
--- a/src/backend/access/index/indexam.c
+++ b/src/backend/access/index/indexam.c
@@ -26,6 +26,7 @@
  *		index_vacuum_cleanup	- post-deletion cleanup of an index
  *		index_getprocid - get a support procedure OID
  *		index_getprocinfo - get a support procedure's lookup info
+ *      index_check_constraint - check index constraints
  *
  * NOTES
  *		This file contains the index_ routines which used
@@ -64,9 +65,13 @@
 
 #include "access/relscan.h"
 #include "access/transam.h"
+#include "miscadmin.h"
 #include "pgstat.h"
 #include "storage/bufmgr.h"
 #include "storage/lmgr.h"
+#include "storage/lwlock.h"
+#include "storage/procarray.h"
+#include "utils/lsyscache.h"
 #include "utils/relcache.h"
 #include "utils/snapmgr.h"
 #include "utils/tqual.h"
@@ -116,6 +121,19 @@ do { \
 static IndexScanDesc index_beginscan_internal(Relation indexRelation,
 						 int nkeys, ScanKey key);
 
+typedef struct
+{
+	Oid					relid;
+	TransactionId		xid;
+	ItemPointerData		tid;
+} CurrentIndexInsertEntry;
+
+static CurrentIndexInsertEntry *CurrentIndexInsertsTable = NULL;
+
+static bool index_check_constraint_conflict(TupleTableSlot *slot,
+											HeapTuple tup, int2 *heap_attnums,
+											int2 index_natts,
+											Oid *constraint_procs);
 
 /* ----------------------------------------------------------------
  *				   index_ interface functions
@@ -846,3 +864,278 @@ index_getprocinfo(Relation irel,
 
 	return locinfo;
 }
+
+void
+index_check_constraint(Relation heap, Relation index,
+						ItemPointer tid, TupleTableSlot *slot)
+{
+		IndexScanDesc	 index_scan;
+		HeapTuple		 tup;
+		ScanKeyData		*scankeys;
+		int2vector		*constr_strats;
+		Oid				*constr_procs;
+		int				 i;
+		int2			*heap_attnums = index->rd_index->indkey.values;
+		int2			 index_natts  = index->rd_index->indnatts;
+		SnapshotData	 DirtySnapshot;
+		int				 nkeys		  = 0;
+
+		CurrentIndexInsertEntry *MyIndexInsertEntry;
+		CurrentIndexInsertEntry	 potential_conflicts[MaxBackends];
+		int						 n_potential_conflicts = 0;
+
+		/* Find constraint strategy numbers */
+		constr_strats = RelationGetIndexConstraintStrategies(index);
+
+		/* return if no constraint */
+		if (constr_strats == NULL)
+			return;
+
+		/*
+		 * if any of the indexed columns are NULL, the constraint
+		 * is satisfied
+		 */
+		for (i = 0; i < index_natts; i++)
+			if (slot_attisnull(slot, heap_attnums[i]))
+				return;
+
+		/*
+		 * Find the function that tests for a conflict based on the
+		 * strategy number, operator family, and types.
+		 */
+		constr_procs = palloc(sizeof(Oid) * index_natts);
+		for (i = 0; i < index_natts; i++)
+		{
+			/*
+			 * Find the procedure implementing the strategy for the
+			 * index for two arguments both with the type of the
+			 * indexed attribute.
+			 */
+			Oid					oper;
+			Oid					opfamily = index->rd_opfamily[i];
+			StrategyNumber		strategy = constr_strats->values[i];
+			Oid	typeOid	= heap->rd_att->attrs[heap_attnums[i] - 1]->atttypid;
+
+			if (strategy == InvalidStrategy)
+				continue;
+
+			oper = get_opfamily_member(opfamily, typeOid, typeOid, strategy);
+
+			if(OidIsValid(oper))
+				constr_procs[i] = get_opcode(oper);
+			else
+				elog(ERROR, "cannot determine operator for type %d and "
+					 "strategy %d", typeOid, strategy);
+		}
+
+
+		if (CurrentIndexInsertsTable == NULL)
+		{
+			bool found;
+
+			CurrentIndexInsertsTable = (CurrentIndexInsertEntry *)
+				ShmemInitStruct("Current Index Inserts Table",
+								CurrentIndexInsertsShmemSize(), &found);
+			Assert(found);
+		}
+
+		MyIndexInsertEntry = &CurrentIndexInsertsTable[MyBackendId - 1];
+
+		/*
+		 * Check for conflicts with concurrent inserts. These are
+		 * inserts that are actually in-progress now, and have not
+		 * actually been put in the index yet.
+		 */
+
+		/* TODO: this lock should be partitioned by heap relid hash. */
+		LWLockAcquire(IndexConstraintLock, LW_EXCLUSIVE);
+
+		for (i = 0; i < MaxBackends; i++)
+		{
+			CurrentIndexInsertEntry entry = CurrentIndexInsertsTable[i];
+
+			if (RelationGetRelid(heap) == entry.relid &&
+				!TransactionIdIsCurrentTransactionId(entry.xid))
+			{
+				if (TransactionIdIsInProgress(entry.xid))
+				{
+					/* build a list of potential conflicts */
+					potential_conflicts[n_potential_conflicts++] = entry;
+				}
+				else
+					entry.relid = InvalidOid;
+			}
+		}
+
+		MyIndexInsertEntry->relid = heap->rd_id;
+		MyIndexInsertEntry->xid	  = GetCurrentTransactionId();
+		MyIndexInsertEntry->tid	  = *tid;
+
+		LWLockRelease(IndexConstraintLock);
+
+		InitDirtySnapshot(DirtySnapshot);
+
+		for (i = 0; i < n_potential_conflicts; i++)
+		{
+			bool				does_conflict = true;
+			HeapTupleData		tup;
+			Buffer				buffer;
+
+			tup.t_self = potential_conflicts[i].tid;
+			if (!heap_fetch(heap, &DirtySnapshot, &tup, &buffer, false, NULL))
+				continue;
+
+			does_conflict = index_check_constraint_conflict(
+				slot, &tup, heap_attnums, index_natts, constr_procs);
+
+			ReleaseBuffer(buffer);
+
+			if (does_conflict)
+			{
+				CurrentIndexInsertEntry conflict = potential_conflicts[i];
+				if (TransactionIdIsCurrentTransactionId(conflict.xid))
+					elog(ERROR, "conflict detected 1");
+
+				XactLockTableWait(conflict.xid);
+				if (TransactionIdDidCommit(conflict.xid))
+					elog(ERROR, "conflict detected 2");
+			}
+		}
+
+		/*
+		 * Now search the tuples that are actually in the index for
+		 * any violations.
+		 */
+
+		scankeys = palloc(index_natts * sizeof(ScanKeyData));
+		for (i = 0; i < index_natts; i++)
+		{
+			Datum	key_datum;
+			bool	isnull;
+
+			key_datum = slot_getattr(slot, heap_attnums[i], &isnull);
+			Assert(!isnull); /* already checked above */
+
+			if (constr_strats->values[i] == InvalidStrategy)
+				continue;
+
+			ScanKeyInit(&scankeys[nkeys], i + 1, constr_strats->values[i],
+						constr_procs[i], key_datum);
+			nkeys++;
+		}
+
+		/*
+		 * We have to find all tuples, even those not visible
+		 * yet. Other transactions may have inserted many tuples (or
+		 * the transaction might be a prepared transaction), so there
+		 * may be some tuples that are not in the shared memory
+		 * structure and not visible.
+		 */
+		index_scan = index_beginscan(heap, index, &DirtySnapshot, nkeys,
+									 scankeys);
+		while((tup = index_getnext(index_scan, ForwardScanDirection)) != NULL)
+		{
+			if (index_scan->xs_recheck)
+			{
+				if (!index_check_constraint_conflict(
+						slot, tup, heap_attnums, index_natts, constr_procs))
+					continue;
+			}
+
+			/* If the in-progress inserting transaction aborts, proceed. */
+			if (TransactionIdIsValid(DirtySnapshot.xmin))
+			{
+				XactLockTableWait(DirtySnapshot.xmin);
+				if (TransactionIdDidAbort(DirtySnapshot.xmin))
+					continue;
+			}
+
+			/* If the in-progress deleting transaction commits, proceed. */
+			if (TransactionIdIsValid(DirtySnapshot.xmax))
+			{
+				XactLockTableWait(DirtySnapshot.xmax);
+				if (TransactionIdDidCommit(DirtySnapshot.xmax))
+					continue;
+			}
+
+			elog(ERROR, "conflict detected 3");
+		}
+
+		index_endscan(index_scan);
+		pfree(scankeys);
+
+		return;
+}
+
+/*
+ * For each attribute of the index, check for a conflict between the
+ * slot's tuple's value and the tuple's value. Only return true if all
+ * values conflict.
+ */
+static bool
+index_check_constraint_conflict(TupleTableSlot *slot, HeapTuple tup,
+								int2 *heap_attnums, int2 index_natts,
+								Oid* constraint_procs)
+{
+	int i;
+
+	for (i = 0; i < index_natts; i++)
+	{
+		Datum	input_datum;
+		Datum	existing_datum;
+		bool	isnull;
+
+		if (!OidIsValid(constraint_procs[i]))
+			continue;
+
+		input_datum = slot_getattr(slot, heap_attnums[i], &isnull);
+		if (isnull)
+			return false;
+
+		existing_datum = heap_getattr(
+			tup, heap_attnums[i], slot->tts_tupleDescriptor, &isnull);
+		if (isnull)
+			return false;
+
+		if (!DatumGetBool(OidFunctionCall2(constraint_procs[i],
+										   input_datum, existing_datum)))
+			return false;
+	}
+	return true;
+}
+
+/*
+ * GistShmemSize --- report amount of shared memory space needed
+ */
+Size
+CurrentIndexInsertsShmemSize(void)
+{
+	return (sizeof(CurrentIndexInsertEntry) * MaxBackends);
+}
+
+/*
+ * GistShmemInit --- initialize this module's shared memory
+ */
+void
+CurrentIndexInsertsShmemInit(void)
+{
+	int							 i;
+	bool						 found;
+	CurrentIndexInsertEntry		*current_inserts;
+
+	current_inserts = (CurrentIndexInsertEntry *)
+		ShmemInitStruct("Current Index Inserts Table",
+						CurrentIndexInsertsShmemSize(),
+						&found);
+
+	if (!IsUnderPostmaster)
+	{
+		/* Initialize shared memory area */
+		Assert(!found);
+
+		for (i = 0; i < MaxBackends; i++)
+			current_inserts[i].relid = InvalidOid;
+	}
+	else
+		Assert(found);
+}
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index c4e4cab..8a136bb 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -444,6 +444,7 @@ UpdateIndexRelation(Oid indexoid,
 	values[Anum_pg_index_indcheckxmin - 1] = BoolGetDatum(false);
 	/* we set isvalid and isready the same way */
 	values[Anum_pg_index_indisready - 1] = BoolGetDatum(isvalid);
+	nulls[Anum_pg_index_indconstrats - 1] = true;
 	values[Anum_pg_index_indkey - 1] = PointerGetDatum(indkey);
 	values[Anum_pg_index_indclass - 1] = PointerGetDatum(indclass);
 	values[Anum_pg_index_indoption - 1] = PointerGetDatum(indoption);
diff --git a/src/backend/executor/execUtils.c b/src/backend/executor/execUtils.c
index 0ccd862..e8d704c 100644
--- a/src/backend/executor/execUtils.c
+++ b/src/backend/executor/execUtils.c
@@ -1067,6 +1067,15 @@ ExecInsertIndexTuples(TupleTableSlot *slot,
 	econtext->ecxt_scantuple = slot;
 
 	/*
+	 * before actually inserting, check index constraints for each index
+	 */
+	for (i = 0; i < numIndices; i++)
+	{
+		index_check_constraint(heapRelation, relationDescs[i],
+							   tupleid, slot);
+	}
+
+	/*
 	 * for each index, form and insert the index tuple
 	 */
 	for (i = 0; i < numIndices; i++)
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index 3022867..8c04940 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -15,6 +15,7 @@
 #include "postgres.h"
 
 #include "access/clog.h"
+#include "access/genam.h"
 #include "access/heapam.h"
 #include "access/multixact.h"
 #include "access/nbtree.h"
@@ -115,6 +116,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
 		size = add_size(size, BgWriterShmemSize());
 		size = add_size(size, AutoVacuumShmemSize());
 		size = add_size(size, BTreeShmemSize());
+		size = add_size(size, CurrentIndexInsertsShmemSize());
 		size = add_size(size, SyncScanShmemSize());
 #ifdef EXEC_BACKEND
 		size = add_size(size, ShmemBackendArraySize());
@@ -215,6 +217,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
 	 * Set up other modules that need some shared memory space
 	 */
 	BTreeShmemInit();
+	CurrentIndexInsertsShmemInit();
 	SyncScanShmemInit();
 
 #ifdef EXEC_BACKEND
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 29976e7..24583d3 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -3284,6 +3284,22 @@ RelationGetIndexAttrBitmap(Relation relation)
 	return indexattrs;
 }
 
+int2vector *
+RelationGetIndexConstraintStrategies(Relation relation)
+{
+	bool		isnull;
+	Datum		constraint_strategies;
+
+	constraint_strategies = heap_getattr(relation->rd_indextuple,
+										 Anum_pg_index_indconstrats,
+										 GetPgIndexDescriptor(),
+										 &isnull);
+	if (isnull)
+		return NULL;
+
+	return (int2vector *) DatumGetPointer(constraint_strategies);
+
+}
 
 /*
  *	load_relcache_init_file, write_relcache_init_file
diff --git a/src/include/access/genam.h b/src/include/access/genam.h
index a6ac5db..2aad8b4 100644
--- a/src/include/access/genam.h
+++ b/src/include/access/genam.h
@@ -16,6 +16,8 @@
 
 #include "access/sdir.h"
 #include "access/skey.h"
+#include "access/xact.h"
+#include "executor/tuptable.h"
 #include "nodes/tidbitmap.h"
 #include "storage/buf.h"
 #include "storage/lock.h"
@@ -129,6 +131,13 @@ extern RegProcedure index_getprocid(Relation irel, AttrNumber attnum,
 				uint16 procnum);
 extern FmgrInfo *index_getprocinfo(Relation irel, AttrNumber attnum,
 				  uint16 procnum);
+extern void index_check_constraint(Relation heapRelation,
+								   Relation indexRelation,
+								   ItemPointer tid,
+								   TupleTableSlot *slot);
+
+extern Size CurrentIndexInsertsShmemSize(void);
+extern void CurrentIndexInsertsShmemInit(void);
 
 /*
  * index access method support routines (in genam.c)
diff --git a/src/include/catalog/pg_attribute.h b/src/include/catalog/pg_attribute.h
index eaa405f..413cf0a 100644
--- a/src/include/catalog/pg_attribute.h
+++ b/src/include/catalog/pg_attribute.h
@@ -477,6 +477,7 @@ DATA(insert ( 1259 tableoid			26 0  4  -7 0 -1 -1 t p i t f f t 0 _null_));
 { 0, {"indclass"},			30, -1, -1, 11, 1, -1, -1, false, 'p', 'i', true, false, false, true, 0, { 0 } }, \
 { 0, {"indoption"},			22, -1, -1, 12, 1, -1, -1, false, 'p', 'i', true, false, false, true, 0, { 0 } }, \
 { 0, {"indexprs"},			25, -1, -1, 13, 0, -1, -1, false, 'x', 'i', false, false, false, true, 0, { 0 } }, \
-{ 0, {"indpred"},			25, -1, -1, 14, 0, -1, -1, false, 'x', 'i', false, false, false, true, 0, { 0 } }
+{ 0, {"indpred"},			25, -1, -1, 14, 0, -1, -1, false, 'x', 'i', false, false, false, true, 0, { 0 } }, \
+{ 0, {"indconstrats"},		22, -1, -1, 15, 1, -1, -1, false, 'p', 'i', false, false, false, true, 0, { 0 } }
 
 #endif   /* PG_ATTRIBUTE_H */
diff --git a/src/include/catalog/pg_index.h b/src/include/catalog/pg_index.h
index 19069db..9ab905c 100644
--- a/src/include/catalog/pg_index.h
+++ b/src/include/catalog/pg_index.h
@@ -49,6 +49,7 @@ CATALOG(pg_index,2610) BKI_WITHOUT_OIDS
 								 * each zero entry in indkey[] */
 	text		indpred;		/* expression tree for predicate, if a partial
 								 * index; else NULL */
+	int2vector	indconstrats;	/* index constraint strategies */
 } FormData_pg_index;
 
 /* ----------------
@@ -62,7 +63,7 @@ typedef FormData_pg_index *Form_pg_index;
  *		compiler constants for pg_index
  * ----------------
  */
-#define Natts_pg_index					14
+#define Natts_pg_index					15
 #define Anum_pg_index_indexrelid		1
 #define Anum_pg_index_indrelid			2
 #define Anum_pg_index_indnatts			3
@@ -77,6 +78,7 @@ typedef FormData_pg_index *Form_pg_index;
 #define Anum_pg_index_indoption			12
 #define Anum_pg_index_indexprs			13
 #define Anum_pg_index_indpred			14
+#define Anum_pg_index_indconstrats		15
 
 /*
  * Index AMs that support ordered scans must support these two indoption
diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h
index e389c61..9e6f93e 100644
--- a/src/include/storage/lwlock.h
+++ b/src/include/storage/lwlock.h
@@ -63,6 +63,7 @@ typedef enum LWLockId
 	TwoPhaseStateLock,
 	TablespaceCreateLock,
 	BtreeVacuumLock,
+	IndexConstraintLock,
 	AddinShmemInitLock,
 	AutovacuumLock,
 	AutovacuumScheduleLock,
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index 7d4d914..305d1d2 100644
--- a/src/include/utils/relcache.h
+++ b/src/include/utils/relcache.h
@@ -43,6 +43,7 @@ extern Oid	RelationGetOidIndex(Relation relation);
 extern List *RelationGetIndexExpressions(Relation relation);
 extern List *RelationGetIndexPredicate(Relation relation);
 extern Bitmapset *RelationGetIndexAttrBitmap(Relation relation);
+extern int2vector *RelationGetIndexConstraintStrategies(Relation relation);
 
 extern void RelationSetIndexList(Relation relation,
 					 List *indexIds, Oid oidIndex);