git diff --patience

diff --git a/src/include/storage/predicate.h b/src/include/storage/predicate.h
index d071c21..7dcc2af 100644
--- a/src/include/storage/predicate.h
+++ b/src/include/storage/predicate.h
@@ -1,7 +1,7 @@
 /*-------------------------------------------------------------------------
  *
  * predicate.h
- *	  POSTGRES predicate locking definitions.
+ *	  POSTGRES public predicate locking definitions.
  *
  *
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
@@ -14,137 +14,15 @@
 #ifndef PREDICATE_H
 #define PREDICATE_H
 
-#include "access/htup.h"
-#include "storage/lock.h"
 #include "utils/relcache.h"
 #include "utils/snapshot.h"
 
-/* GUC variables */
-extern int	max_predicate_locks_per_xact;
-
-/*
- * The SERIALIZABLEXACTTAG struct identifies a serializable transaction.
- */
-typedef struct SERIALIZABLEXACTTAG
-{
-	VirtualTransactionId vxid;	/* We always have one of these. */
-} SERIALIZABLEXACTTAG;
-
-/*
- * Information needed for each serializable database transaction to support SSI techniques.
- * TODO SSI: Should inConflict and outConflict be lists?  That would allow us to reduce
- *			 false positives, *and* would allow us to guarantee that an immediate retry
- *			 of a transaction would never fail on the exact same conflicts.
- *			 The RAM doesn't look like it would be the limiting factor, but CPU time might
- *			 be -- we should have baseline benchmarks before attempting this.
- */
-typedef struct SERIALIZABLEXACT
-{
-	/* hash key */
-	SERIALIZABLEXACTTAG tag;
-
-	/* data */
-	struct SERIALIZABLEXACT *outConflict;		/* ptr to write transaction
-												 * whose data we couldn't
-												 * read. invalid means no
-												 * conflict; self-reference
-												 * means multiple or
-												 * committed. */
-	struct SERIALIZABLEXACT *inConflict;		/* ptr to read transaction
-												 * which couldn't see our
-												 * write. invalid means no
-												 * conflict; self-reference
-												 * means multiple or
-												 * committed. */
-	TransactionId topXid;		/* top level xid for the transaction, if one
-								 * exists */
-	TransactionId finishedBefore;		/* invalid means still running; else
-										 * the struct expires when no tags <
-										 * this. */
-	TransactionId xmin;			/* the transaction's snapshot xmin */
-	SHM_QUEUE	predicateLocks; /* list of associated PREDICATELOCK objects */
-	SHM_QUEUE	xids;			/* list of associated SERIALIZABLEXID objects */
-	SHM_QUEUE	finishedLink;	/* list link in
-								 * FinishedSerializableTransactions */
-	bool		rolledBack;		/* ignore conflicts when true; allows deferred
-								 * cleanup */
-} SERIALIZABLEXACT;
-
-
-typedef enum PredicateLockTargetType
-{
-	PREDLOCKTAG_RELATION,
-	PREDLOCKTAG_PAGE,
-	PREDLOCKTAG_TUPLE
-	/* TODO Other types may be needed for index locking */
-}	PredicateLockTargetType;
 
 /*
- * The PREDICATELOCKTARGETTAG struct is defined to fit into 16
- * bytes with no padding.  Note that this would need adjustment if we were
- * to widen Oid or BlockNumber to more than 32 bits.
- */
-typedef struct PREDICATELOCKTARGETTAG
-{
-	uint32		locktag_field1; /* a 32-bit ID field */
-	uint32		locktag_field2; /* a 32-bit ID field */
-	uint32		locktag_field3; /* a 32-bit ID field */
-	uint16		locktag_field4; /* a 16-bit ID field */
-	uint16		locktag_field5; /* a 16-bit ID field */
-} PREDICATELOCKTARGETTAG;
-
-/*
- * These macros define how we map logical IDs of lockable objects into
- * the physical fields of PREDICATELOCKTARGETTAG.	Use these to set up values,
- * rather than accessing the fields directly.  Note multiple eval of target!
- *
- * TODO SSI: If we always use the same fields for the same type of value,
- * we should rename these.	Holding off until it's clear there are no exceptions.
- * Since indexes are relations with blocks and tuples, it's looking likely that
- * the rename will be possible.  If not, we may need to divide the last field
- * and use part of it for a target type, so that we know how to interpret the
- * data..
+ * GUC variables
  */
-#define SET_PREDICATELOCKTARGETTAG_RELATION(locktag,dboid,reloid) \
-	((locktag).locktag_field1 = (dboid), \
-	 (locktag).locktag_field2 = (reloid), \
-	 (locktag).locktag_field3 = InvalidBlockNumber, \
-	 (locktag).locktag_field4 = InvalidOffsetNumber, \
-	 (locktag).locktag_field5 = 0)
-
-#define SET_PREDICATELOCKTARGETTAG_PAGE(locktag,dboid,reloid,blocknum) \
-	((locktag).locktag_field1 = (dboid), \
-	 (locktag).locktag_field2 = (reloid), \
-	 (locktag).locktag_field3 = (blocknum), \
-	 (locktag).locktag_field4 = InvalidOffsetNumber, \
-	 (locktag).locktag_field5 = 0)
-
-#define SET_PREDICATELOCKTARGETTAG_TUPLE(locktag,dboid,reloid,blocknum,offnum) \
-	((locktag).locktag_field1 = (dboid), \
-	 (locktag).locktag_field2 = (reloid), \
-	 (locktag).locktag_field3 = (blocknum), \
-	 (locktag).locktag_field4 = (offnum), \
-	 (locktag).locktag_field5 = 0)
-
-#define GET_PREDICATELOCKTARGETTAG_DB(locktag) \
-	((locktag).locktag_field1)
-#define GET_PREDICATELOCKTARGETTAG_RELATION(locktag) \
-	((locktag).locktag_field2)
-#define GET_PREDICATELOCKTARGETTAG_PAGE(locktag) \
-	((locktag).locktag_field3)
-#define GET_PREDICATELOCKTARGETTAG_OFFSET(locktag) \
-	((locktag).locktag_field4)
-#define GET_PREDICATELOCKTARGETTAG_TYPE(locktag)							 \
-	(((locktag).locktag_field4 != InvalidOffsetNumber) ? PREDLOCKTAG_TUPLE : \
-	 (((locktag).locktag_field3 != InvalidBlockNumber) ? PREDLOCKTAG_PAGE :   \
-	  PREDLOCKTAG_RELATION))
+extern int	max_predicate_locks_per_xact;
 
-typedef struct PredicateLockData
-{
-	int			nelements;
-	PREDICATELOCKTARGETTAG *locktags;
-	SERIALIZABLEXACT *xacts;
-} PredicateLockData;
 
 /*
  * function prototypes
@@ -155,7 +33,6 @@ extern void InitPredicateLocks(void);
 extern Size PredicateLockShmemSize(void);
 
 /* predicate lock reporting */
-extern PredicateLockData *GetPredicateLockStatusData(void);
 extern bool PageIsPredicateLocked(const Relation relation, const BlockNumber blkno);
 
 /* predicate lock maintenance */

diff --git a/src/include/storage/predicate.h b/src/include/storage/predicate.h
index d071c21..7dcc2af 100644
--- a/src/include/storage/predicate.h
+++ b/src/include/storage/predicate.h
@@ -1,7 +1,7 @@
 /*-------------------------------------------------------------------------
  *
  * predicate.h
- *	  POSTGRES predicate locking definitions.
+ *	  POSTGRES public predicate locking definitions.
  *
  *
  * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
@@ -14,137 +14,15 @@
 #ifndef PREDICATE_H
 #define PREDICATE_H
 
-#include "access/htup.h"
-#include "storage/lock.h"
 #include "utils/relcache.h"
 #include "utils/snapshot.h"
 
-/* GUC variables */
+
+/*
+ * GUC variables
+ */
 extern int	max_predicate_locks_per_xact;
 
-/*
- * The SERIALIZABLEXACTTAG struct identifies a serializable transaction.
- */
-typedef struct SERIALIZABLEXACTTAG
-{
-	VirtualTransactionId vxid;	/* We always have one of these. */
-} SERIALIZABLEXACTTAG;
-
-/*
- * Information needed for each serializable database transaction to support SSI techniques.
- * TODO SSI: Should inConflict and outConflict be lists?  That would allow us to reduce
- *			 false positives, *and* would allow us to guarantee that an immediate retry
- *			 of a transaction would never fail on the exact same conflicts.
- *			 The RAM doesn't look like it would be the limiting factor, but CPU time might
- *			 be -- we should have baseline benchmarks before attempting this.
- */
-typedef struct SERIALIZABLEXACT
-{
-	/* hash key */
-	SERIALIZABLEXACTTAG tag;
-
-	/* data */
-	struct SERIALIZABLEXACT *outConflict;		/* ptr to write transaction
-												 * whose data we couldn't
-												 * read. invalid means no
-												 * conflict; self-reference
-												 * means multiple or
-												 * committed. */
-	struct SERIALIZABLEXACT *inConflict;		/* ptr to read transaction
-												 * which couldn't see our
-												 * write. invalid means no
-												 * conflict; self-reference
-												 * means multiple or
-												 * committed. */
-	TransactionId topXid;		/* top level xid for the transaction, if one
-								 * exists */
-	TransactionId finishedBefore;		/* invalid means still running; else
-										 * the struct expires when no tags <
-										 * this. */
-	TransactionId xmin;			/* the transaction's snapshot xmin */
-	SHM_QUEUE	predicateLocks; /* list of associated PREDICATELOCK objects */
-	SHM_QUEUE	xids;			/* list of associated SERIALIZABLEXID objects */
-	SHM_QUEUE	finishedLink;	/* list link in
-								 * FinishedSerializableTransactions */
-	bool		rolledBack;		/* ignore conflicts when true; allows deferred
-								 * cleanup */
-} SERIALIZABLEXACT;
-
-
-typedef enum PredicateLockTargetType
-{
-	PREDLOCKTAG_RELATION,
-	PREDLOCKTAG_PAGE,
-	PREDLOCKTAG_TUPLE
-	/* TODO Other types may be needed for index locking */
-}	PredicateLockTargetType;
-
-/*
- * The PREDICATELOCKTARGETTAG struct is defined to fit into 16
- * bytes with no padding.  Note that this would need adjustment if we were
- * to widen Oid or BlockNumber to more than 32 bits.
- */
-typedef struct PREDICATELOCKTARGETTAG
-{
-	uint32		locktag_field1; /* a 32-bit ID field */
-	uint32		locktag_field2; /* a 32-bit ID field */
-	uint32		locktag_field3; /* a 32-bit ID field */
-	uint16		locktag_field4; /* a 16-bit ID field */
-	uint16		locktag_field5; /* a 16-bit ID field */
-} PREDICATELOCKTARGETTAG;
-
-/*
- * These macros define how we map logical IDs of lockable objects into
- * the physical fields of PREDICATELOCKTARGETTAG.	Use these to set up values,
- * rather than accessing the fields directly.  Note multiple eval of target!
- *
- * TODO SSI: If we always use the same fields for the same type of value,
- * we should rename these.	Holding off until it's clear there are no exceptions.
- * Since indexes are relations with blocks and tuples, it's looking likely that
- * the rename will be possible.  If not, we may need to divide the last field
- * and use part of it for a target type, so that we know how to interpret the
- * data..
- */
-#define SET_PREDICATELOCKTARGETTAG_RELATION(locktag,dboid,reloid) \
-	((locktag).locktag_field1 = (dboid), \
-	 (locktag).locktag_field2 = (reloid), \
-	 (locktag).locktag_field3 = InvalidBlockNumber, \
-	 (locktag).locktag_field4 = InvalidOffsetNumber, \
-	 (locktag).locktag_field5 = 0)
-
-#define SET_PREDICATELOCKTARGETTAG_PAGE(locktag,dboid,reloid,blocknum) \
-	((locktag).locktag_field1 = (dboid), \
-	 (locktag).locktag_field2 = (reloid), \
-	 (locktag).locktag_field3 = (blocknum), \
-	 (locktag).locktag_field4 = InvalidOffsetNumber, \
-	 (locktag).locktag_field5 = 0)
-
-#define SET_PREDICATELOCKTARGETTAG_TUPLE(locktag,dboid,reloid,blocknum,offnum) \
-	((locktag).locktag_field1 = (dboid), \
-	 (locktag).locktag_field2 = (reloid), \
-	 (locktag).locktag_field3 = (blocknum), \
-	 (locktag).locktag_field4 = (offnum), \
-	 (locktag).locktag_field5 = 0)
-
-#define GET_PREDICATELOCKTARGETTAG_DB(locktag) \
-	((locktag).locktag_field1)
-#define GET_PREDICATELOCKTARGETTAG_RELATION(locktag) \
-	((locktag).locktag_field2)
-#define GET_PREDICATELOCKTARGETTAG_PAGE(locktag) \
-	((locktag).locktag_field3)
-#define GET_PREDICATELOCKTARGETTAG_OFFSET(locktag) \
-	((locktag).locktag_field4)
-#define GET_PREDICATELOCKTARGETTAG_TYPE(locktag)							 \
-	(((locktag).locktag_field4 != InvalidOffsetNumber) ? PREDLOCKTAG_TUPLE : \
-	 (((locktag).locktag_field3 != InvalidBlockNumber) ? PREDLOCKTAG_PAGE :   \
-	  PREDLOCKTAG_RELATION))
-
-typedef struct PredicateLockData
-{
-	int			nelements;
-	PREDICATELOCKTARGETTAG *locktags;
-	SERIALIZABLEXACT *xacts;
-} PredicateLockData;
 
 /*
  * function prototypes
@@ -155,7 +33,6 @@ extern void InitPredicateLocks(void);
 extern Size PredicateLockShmemSize(void);
 
 /* predicate lock reporting */
-extern PredicateLockData *GetPredicateLockStatusData(void);
 extern bool PageIsPredicateLocked(const Relation relation, const BlockNumber blkno);
 
 /* predicate lock maintenance */

diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 788464e..432a4aa 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -1931,6 +1931,7 @@ transformAlterTableStmt(AlterTableStmt *stmt, const char *queryString)
 	List	   *newcmds = NIL;
 	bool		skipValidation = true;
 	AlterTableCmd *newcmd;
+	Oid	index_oid = InvalidOid;
 
 	/*
 	 * We must not scribble on the passed-in AlterTableStmt, so copy it. (This
@@ -2047,35 +2048,152 @@ transformAlterTableStmt(AlterTableStmt *stmt, const char *queryString)
 
 	transformFKConstraints(pstate, &cxt, skipValidation, true);
 
-	/* TODO: look at cxt->pkey->options and see if it has 'USING INDEX'; if yes
-			then cook-up an ALTER TABLE DROP CONSTRAINT for table's primary key,
-			and push into newcmds
+	/*
+	 * If the PRIMARY KEY clasue has WITH INDEX option set, then prepare to use
+	 * that index as the primary key.
+	 */
+	if (cxt.pkey)
+	foreach(l, cxt.pkey->options)
+	{
+		DefElem		*def = (DefElem*)lfirst(l);
+		Oid			pkey_oid;
+		char		*index_name;
+		char		*oid_string;
+		List		*index_name_list;
+		RangeVar	*index_rv;
+		Relation	index_rel;
+
+		if (def->defnamespace == NULL && 0 == strcmp(def->defname, "index"))
+		{
+			if (!(def->defaction == DEFELEM_UNSPEC || def->defaction == DEFELEM_SET))
+				elog(ERROR, "index option for a primary key has a syntax error." );
+		}
+		else
+			continue;
+
+		/*
+		 * If we don't do this, WITH INDEX will get to DefineIndex(), and it will
+		 * throw a fit.
+		 */
+		if (index_oid != InvalidOid)
+			elog(ERROR, "only one WITH INDEX option can be specified for a primary key.");
+
+		if (!IsA(def->arg, String))
+			elog(ERROR, "WITH INDEX option for primary key should be a string value");
+
+		index_name = strVal(def->arg);
 
-	Check if cxt->pkey->options has a 'USING INDEX' element
-		take an exclusive lock on that index
+		index_name_list = stringToQualifiedNameList(index_name);
 
-		Does this table have a primary key
-			check if index mentioned in cxt->pkey matches that PKEY definition,
-				Does column list match
-				Does index type metch (BTree for now)
-				Do they have the same owner
+		/* TODO: Should we assert that this is not a qualified name? */
+		index_rv = makeRangeVarFromNameList(index_name_list);
 
-		Append a new command to newcmds to drop the PKey constraint
-			use 'rel' variable to get primary key's OID ( modify and reuse relationHasPrimaryKey() )
-			use relation_open() to get pkey's relation
-			use the returned Relation->rd_rel->relname to build DROP CONSTRAINT command
-			set missingok member of the command so that this would work even if there was already a DROP CONSTRAINT for the PKey.
-			push this command to newcmds
+		index_rel = relation_openrv(index_rv, AccessExclusiveLock);
 
-		Chenge the 'USING INDEX' element, and replace index name in Value* to have decimal representation of index's OID.
-			This will be used by ATExecAddIndex().
-			string = DatumGetCString(DirectFunctionCall1(oidout,
-												ObjectIdGetDatum(tupleOid)));
+		index_oid = RelationGetRelid(index_rel);
+
+		oid_string = DatumGetCString(DirectFunctionCall1(oidout,
+												ObjectIdGetDatum(index_oid)));
+
+		/*
+		 * Replace index name with its Oid::cstring; ATAddIndex() will use this
+		 * OID for the primary key.
+		 */
+		def->arg = (Node*)makeString(oid_string);
+
+		/* set index name in the statement, to affect the constraint name */
+		cxt.pkey->idxname = pstrdup(strVal(llast(index_name_list)));
+
+		pkey_oid = relationHasPrimaryKey(rel);
+
+		if (pkey_oid != InvalidOid)
+		{
+			HeapTuple		pkey_tuple;
+			HeapTuple		index_tuple;
+			Form_pg_index	pkey_form;
+			Form_pg_index	index_form;
+			AlterTableCmd	*at_cmd;
+			int2			i;
+			Relation		pkey_rel;
+
+			pkey_rel = relation_open(pkey_oid, AccessExclusiveLock);
+
+			/*
+			 * Check pg_class->relam to see if the index type match. As of now, only
+			 * BTree indexes are used to implement primary keys, but it doesn't hurt
+			 * to be future proof.
+			 */
+			if (pkey_rel->rd_rel->relam != index_rel->rd_rel->relam)
+				elog(ERROR, "index type of WITH INDEX argument and that of the primary key are not the same.");
 
-			We need decimal representation of OID since int32 to Oid mapping may not be future proof.
-				TODO check with -hackers
+			if (pkey_rel->rd_rel->relowner != index_rel->rd_rel->relowner)
+				elog(ERROR, "owner of WITH INDEX argument and that of the primary key are not same.");
 
-  */
+			if (pkey_rel->rd_rel->relnamespace != index_rel->rd_rel->relnamespace)
+				elog(ERROR, "index in WITH INDEX argument is not in the correct namespace.");
+
+			pkey_tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(pkey_oid));
+			if (!HeapTupleIsValid(pkey_tuple))
+				elog(ERROR, "cache lookup failed for index %u", pkey_oid);
+			pkey_form = (Form_pg_index) GETSTRUCT(pkey_tuple);
+
+			index_tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(index_oid));
+			if (!HeapTupleIsValid(index_tuple))
+				elog(ERROR, "cache lookup failed for index %u", index_oid);
+			index_form = (Form_pg_index) GETSTRUCT(index_tuple);
+
+			if (!index_form->indisvalid)
+				elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not valid.");
+
+			if (!index_form->indisready)
+				elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not ready.");
+
+			if (!index_form->indisunique)
+				elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not a unique index.");
+
+			/* TODO: Should we check for indcheckxmin ?*/
+
+			if (index_form->indrelid != pkey_form->indrelid)
+				elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not on the same table.");
+
+			if (index_form->indnatts != pkey_form->indnatts)
+				elog(ERROR, "idefinition of index mentioned in the WITH INDEX clause does not match primary key.");
+
+			/*
+			 * Loop over each attribute in the primary key and see if it
+			 * matches the attributes of the index we are replacing it with.
+			 */
+			for (i = 0; i < pkey_form->indnatts; i++)
+			{
+				/* We do not expect primary keys on expressions */
+				Assert(pkey_form->indkey.values[i] != 0);
+
+				if (pkey_form->indkey.values[i] != index_form->indkey.values[i])
+					elog(ERROR, "idefinition of index mentioned in the WITH INDEX clause does not match primary key.");
+
+				if (pkey_form->indclass.values[i] != index_form->indclass.values[i])
+					elog(ERROR, "operator classes of index mentioned in the WITH INDEX clause does not match those of primary key.");
+			}
+
+			ReleaseSysCache(pkey_tuple);
+			ReleaseSysCache(index_tuple);
+
+			at_cmd = makeNode(AlterTableCmd);
+			at_cmd->subtype = AT_DropConstraint;
+			at_cmd->name = pstrdup(NameStr(pkey_rel->rd_rel->relname));
+			at_cmd->behavior = DROP_CASCADE;;
+			at_cmd->missing_ok = true;
+
+			newcmds = lappend(newcmds, at_cmd);
+
+			/* Release the lock so that it can be dropped from cache. */
+			relation_close(pkey_rel, AccessExclusiveLock);
+		}
+
+		relation_close(index_rel, NoLock);
+
+		/* do not break; at the end of the loop. Use this opprtunity to catch multiple 'WITH INDEX' clauses*/
+	}
 
 	/*
 	 * Push any index-creation commands into the ALTER, so that they can be

diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 788464e..432a4aa 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -1931,6 +1931,7 @@ transformAlterTableStmt(AlterTableStmt *stmt, const char *queryString)
 	List	   *newcmds = NIL;
 	bool		skipValidation = true;
 	AlterTableCmd *newcmd;
+	Oid	index_oid = InvalidOid;
 
 	/*
 	 * We must not scribble on the passed-in AlterTableStmt, so copy it. (This
@@ -2047,35 +2048,152 @@ transformAlterTableStmt(AlterTableStmt *stmt, const char *queryString)
 
 	transformFKConstraints(pstate, &cxt, skipValidation, true);
 
-	/* TODO: look at cxt->pkey->options and see if it has 'USING INDEX'; if yes
-			then cook-up an ALTER TABLE DROP CONSTRAINT for table's primary key,
-			and push into newcmds
-
-	Check if cxt->pkey->options has a 'USING INDEX' element
-		take an exclusive lock on that index
-
-		Does this table have a primary key
-			check if index mentioned in cxt->pkey matches that PKEY definition,
-				Does column list match
-				Does index type metch (BTree for now)
-				Do they have the same owner
-
-		Append a new command to newcmds to drop the PKey constraint
-			use 'rel' variable to get primary key's OID ( modify and reuse relationHasPrimaryKey() )
-			use relation_open() to get pkey's relation
-			use the returned Relation->rd_rel->relname to build DROP CONSTRAINT command
-			set missingok member of the command so that this would work even if there was already a DROP CONSTRAINT for the PKey.
-			push this command to newcmds
-
-		Chenge the 'USING INDEX' element, and replace index name in Value* to have decimal representation of index's OID.
-			This will be used by ATExecAddIndex().
-			string = DatumGetCString(DirectFunctionCall1(oidout,
-												ObjectIdGetDatum(tupleOid)));
-
-			We need decimal representation of OID since int32 to Oid mapping may not be future proof.
-				TODO check with -hackers
-
-  */
+	/*
+	 * If the PRIMARY KEY clasue has WITH INDEX option set, then prepare to use
+	 * that index as the primary key.
+	 */
+	if (cxt.pkey)
+	foreach(l, cxt.pkey->options)
+	{
+		DefElem		*def = (DefElem*)lfirst(l);
+		Oid			pkey_oid;
+		char		*index_name;
+		char		*oid_string;
+		List		*index_name_list;
+		RangeVar	*index_rv;
+		Relation	index_rel;
+
+		if (def->defnamespace == NULL && 0 == strcmp(def->defname, "index"))
+		{
+			if (!(def->defaction == DEFELEM_UNSPEC || def->defaction == DEFELEM_SET))
+				elog(ERROR, "index option for a primary key has a syntax error." );
+		}
+		else
+			continue;
+
+		/*
+		 * If we don't do this, WITH INDEX will get to DefineIndex(), and it will
+		 * throw a fit.
+		 */
+		if (index_oid != InvalidOid)
+			elog(ERROR, "only one WITH INDEX option can be specified for a primary key.");
+
+		if (!IsA(def->arg, String))
+			elog(ERROR, "WITH INDEX option for primary key should be a string value");
+
+		index_name = strVal(def->arg);
+
+		index_name_list = stringToQualifiedNameList(index_name);
+
+		/* TODO: Should we assert that this is not a qualified name? */
+		index_rv = makeRangeVarFromNameList(index_name_list);
+
+		index_rel = relation_openrv(index_rv, AccessExclusiveLock);
+
+		index_oid = RelationGetRelid(index_rel);
+
+		oid_string = DatumGetCString(DirectFunctionCall1(oidout,
+												ObjectIdGetDatum(index_oid)));
+
+		/*
+		 * Replace index name with its Oid::cstring; ATAddIndex() will use this
+		 * OID for the primary key.
+		 */
+		def->arg = (Node*)makeString(oid_string);
+
+		/* set index name in the statement, to affect the constraint name */
+		cxt.pkey->idxname = pstrdup(strVal(llast(index_name_list)));
+
+		pkey_oid = relationHasPrimaryKey(rel);
+
+		if (pkey_oid != InvalidOid)
+		{
+			HeapTuple		pkey_tuple;
+			HeapTuple		index_tuple;
+			Form_pg_index	pkey_form;
+			Form_pg_index	index_form;
+			AlterTableCmd	*at_cmd;
+			int2			i;
+			Relation		pkey_rel;
+
+			pkey_rel = relation_open(pkey_oid, AccessExclusiveLock);
+
+			/*
+			 * Check pg_class->relam to see if the index type match. As of now, only
+			 * BTree indexes are used to implement primary keys, but it doesn't hurt
+			 * to be future proof.
+			 */
+			if (pkey_rel->rd_rel->relam != index_rel->rd_rel->relam)
+				elog(ERROR, "index type of WITH INDEX argument and that of the primary key are not the same.");
+
+			if (pkey_rel->rd_rel->relowner != index_rel->rd_rel->relowner)
+				elog(ERROR, "owner of WITH INDEX argument and that of the primary key are not same.");
+
+			if (pkey_rel->rd_rel->relnamespace != index_rel->rd_rel->relnamespace)
+				elog(ERROR, "index in WITH INDEX argument is not in the correct namespace.");
+
+			pkey_tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(pkey_oid));
+			if (!HeapTupleIsValid(pkey_tuple))
+				elog(ERROR, "cache lookup failed for index %u", pkey_oid);
+			pkey_form = (Form_pg_index) GETSTRUCT(pkey_tuple);
+
+			index_tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(index_oid));
+			if (!HeapTupleIsValid(index_tuple))
+				elog(ERROR, "cache lookup failed for index %u", index_oid);
+			index_form = (Form_pg_index) GETSTRUCT(index_tuple);
+
+			if (!index_form->indisvalid)
+				elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not valid.");
+
+			if (!index_form->indisready)
+				elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not ready.");
+
+			if (!index_form->indisunique)
+				elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not a unique index.");
+
+			/* TODO: Should we check for indcheckxmin ?*/
+
+			if (index_form->indrelid != pkey_form->indrelid)
+				elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not on the same table.");
+
+			if (index_form->indnatts != pkey_form->indnatts)
+				elog(ERROR, "idefinition of index mentioned in the WITH INDEX clause does not match primary key.");
+
+			/*
+			 * Loop over each attribute in the primary key and see if it
+			 * matches the attributes of the index we are replacing it with.
+			 */
+			for (i = 0; i < pkey_form->indnatts; i++)
+			{
+				/* We do not expect primary keys on expressions */
+				Assert(pkey_form->indkey.values[i] != 0);
+
+				if (pkey_form->indkey.values[i] != index_form->indkey.values[i])
+					elog(ERROR, "idefinition of index mentioned in the WITH INDEX clause does not match primary key.");
+
+				if (pkey_form->indclass.values[i] != index_form->indclass.values[i])
+					elog(ERROR, "operator classes of index mentioned in the WITH INDEX clause does not match those of primary key.");
+			}
+
+			ReleaseSysCache(pkey_tuple);
+			ReleaseSysCache(index_tuple);
+
+			at_cmd = makeNode(AlterTableCmd);
+			at_cmd->subtype = AT_DropConstraint;
+			at_cmd->name = pstrdup(NameStr(pkey_rel->rd_rel->relname));
+			at_cmd->behavior = DROP_CASCADE;;
+			at_cmd->missing_ok = true;
+
+			newcmds = lappend(newcmds, at_cmd);
+
+			/* Release the lock so that it can be dropped from cache. */
+			relation_close(pkey_rel, AccessExclusiveLock);
+		}
+
+		relation_close(index_rel, NoLock);
+
+		/* do not break; at the end of the loop. Use this opprtunity to catch multiple 'WITH INDEX' clauses*/
+	}
 
 	/*
 	 * Push any index-creation commands into the ALTER, so that they can be