replace_pkey_index.patch

diff --git a/src/backend/bootstrap/bootparse.y b/src/backend/bootstrap/bootparse.y
index 387d43e..fed79ff 100644
--- a/src/backend/bootstrap/bootparse.y
+++ b/src/backend/bootstrap/bootparse.y
@@ -287,7 +287,7 @@ Boot_DeclareIndexStmt:
 								$10,
 								NULL, NIL, NIL,
 								false, false, false, false, false,
-								false, false, true, false, false);
+								false, false, true, false, false, false);
 					do_end();
 				}
 		;
@@ -305,7 +305,7 @@ Boot_DeclareUniqueIndexStmt:
 								$11,
 								NULL, NIL, NIL,
 								true, false, false, false, false,
-								false, false, true, false, false);
+								false, false, true, false, false, false);
 					do_end();
 				}
 		;
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 69946fe..0ab523e 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -512,6 +512,7 @@ UpdateIndexRelation(Oid indexoid,
  * allow_system_table_mods: allow table to be a system catalog
  * skip_build: true to skip the index_build() step for the moment; caller
  *		must do it later (typically via reindex_index())
+ * index_exists: the index already exists, we are here just for formalities.
  * concurrent: if true, do not lock the table against writers.	The index
  *		will be marked "invalid" and the caller must take additional steps
  *		to fix it up.
@@ -535,6 +536,7 @@ index_create(Oid heapRelationId,
 			 bool initdeferred,
 			 bool allow_system_table_mods,
 			 bool skip_build,
+			 bool index_exists,
 			 bool concurrent)
 {
 	Relation	pg_class;
@@ -615,6 +617,7 @@ index_create(Oid heapRelationId,
 	if (shared_relation && tableSpaceId != GLOBALTABLESPACE_OID)
 		elog(ERROR, "shared relations must be placed in pg_global tablespace");
 
+	if (!index_exists)
 	if (get_relname_relid(indexRelationName, namespaceId))
 		ereport(ERROR,
 				(errcode(ERRCODE_DUPLICATE_TABLE),
@@ -648,6 +651,17 @@ index_create(Oid heapRelationId,
 			indexRelationId = GetNewRelFileNode(tableSpaceId, pg_class);
 	}
 
+	if (index_exists)
+	{
+		Assert( skip_build && !IsBootstrapProcessingMode() );
+
+		indexRelation = relation_open(indexRelationId, AccessExclusiveLock);
+
+		/* done with pg_class */
+		heap_close(pg_class, RowExclusiveLock);
+	}
+	else
+	{
 	/*
 	 * create the index relation's relcache entry and physical disk file. (If
 	 * we fail further down, it's the smgr's responsibility to remove the disk
@@ -720,6 +734,7 @@ index_create(Oid heapRelationId,
 						classObjectId, coloptions, isprimary,
 						!deferrable,
 						!concurrent);
+	}
 
 	/*
 	 * Register constraint and dependencies for the index.
@@ -833,7 +848,7 @@ index_create(Oid heapRelationId,
 									 true);
 			}
 		}
-		else
+		else if(!index_exists)
 		{
 			bool		have_simple_col = false;
 
@@ -876,6 +891,8 @@ index_create(Oid heapRelationId,
 			Assert(!initdeferred);
 		}
 
+		if (!index_exists)
+		{
 		/* Store dependency on operator classes */
 		for (i = 0; i < indexInfo->ii_NumIndexAttrs; i++)
 		{
@@ -905,6 +922,7 @@ index_create(Oid heapRelationId,
 											DEPENDENCY_NORMAL,
 											DEPENDENCY_AUTO);
 		}
+		}
 	}
 	else
 	{
@@ -945,6 +963,8 @@ index_create(Oid heapRelationId,
 	}
 	else if (skip_build)
 	{
+		if (!index_exists)
+		{
 		/*
 		 * Caller is responsible for filling the index later on.  However,
 		 * we'd better make sure that the heap relation is correctly marked as
@@ -958,6 +978,7 @@ index_create(Oid heapRelationId,
 						   heapRelation->rd_rel->reltuples);
 		/* Make the above update visible */
 		CommandCounterIncrement();
+		}
 	}
 	else
 	{
diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c
index 435dfdd..96cbce2 100644
--- a/src/backend/catalog/toasting.c
+++ b/src/backend/catalog/toasting.c
@@ -269,7 +269,7 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid, Datum reloptio
 							   rel->rd_rel->reltablespace,
 							   classObjectId, coloptions, (Datum) 0,
 							   true, false, false, false,
-							   true, false, false);
+							   true, false, false, false);
 
 	/*
 	 * Store the toast table's OID in the parent relation's pg_class row
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 2da1c86..11ba888 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -69,7 +69,9 @@ static void ComputeIndexAttrs(IndexInfo *indexInfo,
 static Oid GetIndexOpClass(List *opclass, Oid attrType,
 				char *accessMethodName, Oid accessMethodId);
 static char *ChooseIndexNameAddition(List *colnames);
-static bool relationHasPrimaryKey(Relation rel);
+
+/* TODO: Unable to figure out which header to declare this in */
+Oid relationHasPrimaryKey(Relation rel);
 
 
 /*
@@ -122,6 +124,7 @@ DefineIndex(RangeVar *heapRelation,
 			bool is_alter_table,
 			bool check_rights,
 			bool skip_build,
+			bool index_exists,
 			bool quiet,
 			bool concurrent)
 {
@@ -340,7 +343,7 @@ DefineIndex(RangeVar *heapRelation,
 		 * it's no problem either.
 		 */
 		if (is_alter_table &&
-			relationHasPrimaryKey(rel))
+			relationHasPrimaryKey(rel) != InvalidOid)
 		{
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
@@ -488,6 +491,7 @@ DefineIndex(RangeVar *heapRelation,
 					 isconstraint, deferrable, initdeferred,
 					 allowSystemTableMods,
 					 skip_build || concurrent,
+					 index_exists,
 					 concurrent);
 
 	if (!concurrent)
@@ -1541,10 +1545,11 @@ ChooseIndexColumnNames(List *indexElems)
  *
  *	See whether an existing relation has a primary key.
  */
-static bool
+Oid 
 relationHasPrimaryKey(Relation rel)
 {
-	bool		result = false;
+	Oid			indexoid = InvalidOid;
+	bool		isprimary = false;
 	List	   *indexoidlist;
 	ListCell   *indexoidscan;
 
@@ -1557,21 +1562,22 @@ relationHasPrimaryKey(Relation rel)
 
 	foreach(indexoidscan, indexoidlist)
 	{
-		Oid			indexoid = lfirst_oid(indexoidscan);
 		HeapTuple	indexTuple;
 
+		indexoid = lfirst_oid(indexoidscan);
+
 		indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexoid));
 		if (!HeapTupleIsValid(indexTuple))		/* should not happen */
 			elog(ERROR, "cache lookup failed for index %u", indexoid);
-		result = ((Form_pg_index) GETSTRUCT(indexTuple))->indisprimary;
+		isprimary = ((Form_pg_index) GETSTRUCT(indexTuple))->indisprimary;
 		ReleaseSysCache(indexTuple);
-		if (result)
+		if (isprimary)
 			break;
 	}
 
 	list_free(indexoidlist);
 
-	return result;
+	return isprimary ? indexoid : InvalidOid;
 }
 
 /*
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 2383112..68e916f 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -4559,7 +4559,10 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
 {
 	bool		check_rights;
 	bool		skip_build;
+	bool		index_exists = false;
 	bool		quiet;
+	Oid			index_oid = InvalidOid;
+	ListCell	*l, *prev = NULL;
 
 	Assert(IsA(stmt, IndexStmt));
 
@@ -4570,11 +4573,55 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
 	/* suppress notices when rebuilding existing index */
 	quiet = is_rebuild;
 
+	/* TODO:
+		Do stmt->options have 'WITH INDEX' option set
+			Get OID from the Value* (represented as string)
+				strspn(opr_name_or_oid, "0123456789") == strlen(opr_name_or_oid))
+				result = DatumGetObjectId(DirectFunctionCall1(oidin,
+												CStringGetDatum(opr_name_or_oid)));
+			replace InvalidOid, below, with this OID
+			set 'primary' to on;
+			set skip_build to on
+				TODO Check with -hackers that this is the right thing to do even when tab->newvals is non-null?
+	 */
+
+	foreach(l, stmt->options)
+	{
+		DefElem	*def = (DefElem*)lfirst(l);
+
+		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
+		{
+			prev = l;
+			continue;
+		}
+
+		Assert(strspn(strVal(def->arg), "0123456789") == strlen(strVal(def->arg)));
+		Assert(stmt->primary);
+
+		index_oid = DatumGetObjectId(DirectFunctionCall1(oidin,
+										CStringGetDatum(strVal(def->arg))));
+
+		/* We override the skip_build set above */
+		skip_build = true;
+
+		index_exists = true;
+
+		break;
+	}
+
+	if (l) /* unecessary check, but good for readability */
+		stmt->options = list_delete_cell(stmt->options, l, prev);
+
 	/* The IndexStmt has already been through transformIndexStmt */
 
 	DefineIndex(stmt->relation, /* relation */
 				stmt->idxname,	/* index name */
-				InvalidOid,		/* no predefined OID */
+				index_oid,		/* no predefined OID */
 				stmt->accessMethod,		/* am name */
 				stmt->tableSpace,
 				stmt->indexParams,		/* parameters */
@@ -4589,6 +4636,7 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
 				true,			/* is_alter_table */
 				check_rights,
 				skip_build,
+				index_exists,
 				quiet,
 				false);
 }
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 322e689..ad98928 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,6 +2048,192 @@ transformAlterTableStmt(AlterTableStmt *stmt, const char *queryString)
 
 	transformFKConstraints(pstate, &cxt, skipValidation, true);
 
+	/* TODO: look at cxt->pkey->options and see if it has 'WITH 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 'WITH 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
+				Do the opclasses match
+				Does index type match (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 'WITH INDEX' element, and replace index name in Value* to have decimal representation of index's OID.
+			This will be used by ATExecAddIndex().
+
+		BIG TODO: convert all elog() calls into ereport() and proper errcode()
+					calls and write better messages.
+	  */
+
+	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, then this 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 a non-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 */
+		def->arg = (Node*)makeString(oid_string);
+
+		/* TODO: set index name in the statement, to affect the constraint name */
+		cxt.pkey->idxname = pstrdup(strVal(llast(index_name_list)));
+
+		/* TODO: make sure the index is in the same schema/namespace as the table */
+
+		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;
+
+			/*
+			 * 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.
+			 */
+			pkey_rel = relation_open(pkey_oid, AccessExclusiveLock);
+
+			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.");
+
+			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->indisunique)
+				elog(ERROR, "index mentioned in the WITH INDEX clause of primary key is not a unique index.");
+
+			/* TODO: Should we check for indisready, 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);
+		}
+
+		/* NOw update pg_index tuple to mark this index as indisprimary */
+		{
+			Relation	pg_index;
+			HeapTuple	indexTuple;
+			Form_pg_index indexForm;
+
+			pg_index = heap_open(IndexRelationId, RowExclusiveLock);
+
+			indexTuple = SearchSysCacheCopy1(INDEXRELID,
+											 ObjectIdGetDatum(index_oid));
+			if (!HeapTupleIsValid(indexTuple))
+				elog(ERROR, "cache lookup failed for index %u", index_oid);
+			indexForm = (Form_pg_index) GETSTRUCT(indexTuple);
+
+			indexForm->indisprimary = true;
+			simple_heap_update(pg_index, &indexTuple->t_self, indexTuple);
+			CatalogUpdateIndexes(pg_index, indexTuple);
+
+			heap_freetuple(indexTuple);
+			heap_close(pg_index, RowExclusiveLock);
+		}
+
+		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
 	 * scheduled nicely by tablecmds.c.  Note that tablecmds.c assumes that
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index b9be06f..adae7c6 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -890,6 +890,7 @@ standard_ProcessUtility(Node *parsetree,
 							false,		/* is_alter_table */
 							true,		/* check_rights */
 							false,		/* skip_build */
+							false,		/* index_exists */
 							false,		/* quiet */
 							stmt->concurrent);	/* concurrent */
 			}
diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h
index d336576..3b019cb 100644
--- a/src/include/catalog/index.h
+++ b/src/include/catalog/index.h
@@ -44,6 +44,7 @@ extern Oid index_create(Oid heapRelationId,
 			 bool initdeferred,
 			 bool allow_system_table_mods,
 			 bool skip_build,
+			 bool index_exists,
 			 bool concurrent);
 
 extern void index_drop(Oid indexId);
diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h
index 7edda97..c774710 100644
--- a/src/include/commands/defrem.h
+++ b/src/include/commands/defrem.h
@@ -35,6 +35,7 @@ extern void DefineIndex(RangeVar *heapRelation,
 			bool is_alter_table,
 			bool check_rights,
 			bool skip_build,
+			bool index_exists,
 			bool quiet,
 			bool concurrent);
 extern void ReindexIndex(RangeVar *indexRelation);