Avoid index rebuild for no-rewrite ALTER TABLE .. ALTER TYPE.

Noah Misch.  Review and minor cosmetic changes by me.

19 files changed, 341 insertions, 51 deletions

diff --git a/src/backend/bootstrap/bootparse.y b/src/backend/bootstrap/bootparse.y
index f3e85aa31bb..d0a0e9276e8 100644
--- a/src/backend/bootstrap/bootparse.y
+++ b/src/backend/bootstrap/bootparse.y
@@ -217,6 +217,7 @@ Boot_CreateStmt:
 												   PG_CATALOG_NAMESPACE,
 												   shared_relation ? GLOBALTABLESPACE_OID : 0,
 												   $3,
+												   InvalidOid,
 												   tupdesc,
 												   RELKIND_RELATION,
 												   RELPERSISTENCE_PERMANENT,
@@ -284,6 +285,7 @@ Boot_DeclareIndexStmt:
 					DefineIndex(makeRangeVar(NULL, $6, -1),
 								$3,
 								$4,
+								InvalidOid,
 								$8,
 								NULL,
 								$10,
@@ -302,6 +304,7 @@ Boot_DeclareUniqueIndexStmt:
 					DefineIndex(makeRangeVar(NULL, $7, -1),
 								$4,
 								$5,
+								InvalidOid,
 								$9,
 								NULL,
 								$11,
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index f506b8e40dd..7cf44d1f256 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -229,7 +229,8 @@ SystemAttributeByName(const char *attname, bool relhasoids)
  *		heap_create		- Create an uncataloged heap relation
  *
  *		Note API change: the caller must now always provide the OID
- *		to use for the relation.
+ *		to use for the relation.  The relfilenode may (and, normally,
+ *		should) be left unspecified.
  *
  *		rel->rd_rel is initialized by RelationBuildLocalRelation,
  *		and is mostly zeroes at return.
@@ -240,6 +241,7 @@ heap_create(const char *relname,
 			Oid relnamespace,
 			Oid reltablespace,
 			Oid relid,
+			Oid relfilenode,
 			TupleDesc tupDesc,
 			char relkind,
 			char relpersistence,
@@ -297,6 +299,16 @@ heap_create(const char *relname,
 	}
 
 	/*
+	 * Unless otherwise requested, the physical ID (relfilenode) is initially
+	 * the same as the logical ID (OID).  When the caller did specify a
+	 * relfilenode, it already exists; do not attempt to create it.
+	 */
+	if (OidIsValid(relfilenode))
+		create_storage = false;
+	else
+		relfilenode = relid;
+
+	/*
 	 * Never allow a pg_class entry to explicitly specify the database's
 	 * default tablespace in reltablespace; force it to zero instead. This
 	 * ensures that if the database is cloned with a different default
@@ -315,6 +327,7 @@ heap_create(const char *relname,
 									 relnamespace,
 									 tupDesc,
 									 relid,
+									 relfilenode,
 									 reltablespace,
 									 shared_relation,
 									 mapped_relation,
@@ -1103,6 +1116,7 @@ heap_create_with_catalog(const char *relname,
 							   relnamespace,
 							   reltablespace,
 							   relid,
+							   InvalidOid,
 							   tupdesc,
 							   relkind,
 							   relpersistence,
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 83aae7abada..75b4c146f5f 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -649,6 +649,8 @@ UpdateIndexRelation(Oid indexoid,
  * indexRelationId: normally, pass InvalidOid to let this routine
  *		generate an OID for the index.	During bootstrap this may be
  *		nonzero to specify a preselected OID.
+ * relFileNode: normally, pass InvalidOid to get new storage.  May be
+ *		nonzero to attach an existing valid build.
  * indexInfo: same info executor uses to insert into the index
  * indexColNames: column names to use for index (List of char *)
  * accessMethodObjectId: OID of index AM to use
@@ -674,6 +676,7 @@ Oid
 index_create(Relation heapRelation,
 			 const char *indexRelationName,
 			 Oid indexRelationId,
+			 Oid relFileNode,
 			 IndexInfo *indexInfo,
 			 List *indexColNames,
 			 Oid accessMethodObjectId,
@@ -813,6 +816,7 @@ index_create(Relation heapRelation,
 								namespaceId,
 								tableSpaceId,
 								indexRelationId,
+								relFileNode,
 								indexTupDesc,
 								RELKIND_INDEX,
 								relpersistence,
diff --git a/src/backend/catalog/storage.c b/src/backend/catalog/storage.c
index 57987be2c0a..d0273611703 100644
--- a/src/backend/catalog/storage.c
+++ b/src/backend/catalog/storage.c
@@ -206,10 +206,13 @@ RelationDropStorage(Relation rel)
  * The relation mapper fixes this by telling us to not delete such relations
  * after all as part of its commit.
  *
+ * We also use this to reuse an old build of an index during ALTER TABLE, this
+ * time removing the delete-at-commit entry.
+ *
  * No-op if the relation is not among those scheduled for deletion.
  */
 void
-RelationPreserveStorage(RelFileNode rnode)
+RelationPreserveStorage(RelFileNode rnode, bool atCommit)
 {
 	PendingRelDelete *pending;
 	PendingRelDelete *prev;
@@ -219,11 +222,9 @@ RelationPreserveStorage(RelFileNode rnode)
 	for (pending = pendingDeletes; pending != NULL; pending = next)
 	{
 		next = pending->next;
-		if (RelFileNodeEquals(rnode, pending->relnode))
+		if (RelFileNodeEquals(rnode, pending->relnode)
+			&& pending->atCommit == atCommit)
 		{
-			/* we should only find delete-on-abort entries, else trouble */
-			if (pending->atCommit)
-				elog(ERROR, "cannot preserve a delete-on-commit relation");
 			/* unlink and delete list entry */
 			if (prev)
 				prev->next = next;
diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c
index ce082fd2c25..a09a3adb20d 100644
--- a/src/backend/catalog/toasting.c
+++ b/src/backend/catalog/toasting.c
@@ -274,7 +274,7 @@ create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid, Datum reloptio
 	coloptions[0] = 0;
 	coloptions[1] = 0;
 
-	index_create(toast_rel, toast_idxname, toastIndexOid,
+	index_create(toast_rel, toast_idxname, toastIndexOid, InvalidOid,
 				 indexInfo,
 				 list_make2("chunk_id", "chunk_seq"),
 				 BTREE_AM_OID,
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 50248540816..16eae204ab3 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -73,6 +73,198 @@ static char *ChooseIndexNameAddition(List *colnames);
 
 
 /*
+ * CheckIndexCompatible
+ *		Determine whether an existing index definition is compatible with a
+ *		prospective index definition, such that the existing index storage
+ *		could become the storage of the new index, avoiding a rebuild.
+ *
+ * 'heapRelation': the relation the index would apply to.
+ * 'accessMethodName': name of the AM to use.
+ * 'attributeList': a list of IndexElem specifying columns and expressions
+ *		to index on.
+ * 'exclusionOpNames': list of names of exclusion-constraint operators,
+ *		or NIL if not an exclusion constraint.
+ *
+ * This is tailored to the needs of ALTER TABLE ALTER TYPE, which recreates
+ * any indexes that depended on a changing column from their pg_get_indexdef
+ * or pg_get_constraintdef definitions.  We omit some of the sanity checks of
+ * DefineIndex.  We assume that the old and new indexes have the same number
+ * of columns and that if one has an expression column or predicate, both do.
+ * Errors arising from the attribute list still apply.
+ *
+ * Most column type changes that can skip a table rewrite will not invalidate
+ * indexes.  For btree and hash indexes, we assume continued validity when
+ * each column of an index would have the same operator family before and
+ * after the change.  Since we do not document a contract for GIN or GiST
+ * operator families, we require an exact operator class match for them and
+ * for any other access methods.
+ *
+ * DefineIndex always verifies that each exclusion operator shares an operator
+ * family with its corresponding index operator class.  For access methods
+ * having no operator family contract, confirm that the old and new indexes
+ * use the exact same exclusion operator.  For btree and hash, there's nothing
+ * more to check.
+ *
+ * We do not yet implement a test to verify compatibility of expression
+ * columns or predicates, so assume any such index is incompatible.
+ */
+bool
+CheckIndexCompatible(Oid oldId,
+					 RangeVar *heapRelation,
+					 char *accessMethodName,
+					 List *attributeList,
+					 List *exclusionOpNames)
+{
+	bool		isconstraint;
+	Oid		   *collationObjectId;
+	Oid		   *classObjectId;
+	Oid			accessMethodId;
+	Oid			relationId;
+	HeapTuple	tuple;
+	Form_pg_am	accessMethodForm;
+	bool		amcanorder;
+	RegProcedure amoptions;
+	int16	   *coloptions;
+	IndexInfo  *indexInfo;
+	int			numberOfAttributes;
+	int			old_natts;
+	bool		isnull;
+	bool		family_am;
+	bool		ret = true;
+	oidvector  *old_indclass;
+	oidvector  *old_indcollation;
+	int			i;
+	Datum		d;
+
+	/* Caller should already have the relation locked in some way. */
+	relationId = RangeVarGetRelid(heapRelation, NoLock, false, false);
+	/*
+	 * We can pretend isconstraint = false unconditionally.  It only serves to
+	 * decide the text of an error message that should never happen for us.
+	 */
+	isconstraint = false;
+
+	numberOfAttributes = list_length(attributeList);
+	Assert(numberOfAttributes > 0);
+	Assert(numberOfAttributes <= INDEX_MAX_KEYS);
+
+	/* look up the access method */
+	tuple = SearchSysCache1(AMNAME, PointerGetDatum(accessMethodName));
+	if (!HeapTupleIsValid(tuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("access method \"%s\" does not exist",
+						accessMethodName)));
+	accessMethodId = HeapTupleGetOid(tuple);
+	accessMethodForm = (Form_pg_am) GETSTRUCT(tuple);
+	amcanorder = accessMethodForm->amcanorder;
+	amoptions = accessMethodForm->amoptions;
+	ReleaseSysCache(tuple);
+
+	/*
+	 * Compute the operator classes, collations, and exclusion operators
+	 * for the new index, so we can test whether it's compatible with the
+	 * existing one.  Note that ComputeIndexAttrs might fail here, but that's
+	 * OK: DefineIndex would have called this function with the same arguments
+	 * later on, and it would have failed then anyway.
+	 */
+	indexInfo = makeNode(IndexInfo);
+	indexInfo->ii_Expressions = NIL;
+	indexInfo->ii_ExpressionsState = NIL;
+	indexInfo->ii_PredicateState = NIL;
+	indexInfo->ii_ExclusionOps = NULL;
+	indexInfo->ii_ExclusionProcs = NULL;
+	indexInfo->ii_ExclusionStrats = NULL;
+	collationObjectId = (Oid *) palloc(numberOfAttributes * sizeof(Oid));
+	classObjectId = (Oid *) palloc(numberOfAttributes * sizeof(Oid));
+	coloptions = (int16 *) palloc(numberOfAttributes * sizeof(int16));
+	ComputeIndexAttrs(indexInfo, collationObjectId, classObjectId,
+					  coloptions, attributeList,
+					  exclusionOpNames, relationId,
+					  accessMethodName, accessMethodId,
+					  amcanorder, isconstraint);
+
+
+	/* Get the soon-obsolete pg_index tuple. */
+	tuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(oldId));
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for index %u", oldId);
+
+	/* We don't assess expressions or predicates; assume incompatibility. */
+	if (!(heap_attisnull(tuple, Anum_pg_index_indpred) &&
+		  heap_attisnull(tuple, Anum_pg_index_indexprs)))
+	{
+		ReleaseSysCache(tuple);
+		return false;
+	}
+
+	/*
+	 * If the old and new operator class of any index column differ in
+	 * operator family or collation, regard the old index as incompatible.
+	 * For access methods other than btree and hash, a family match has no
+	 * defined meaning; require an exact operator class match.
+	 */
+	old_natts = ((Form_pg_index) GETSTRUCT(tuple))->indnatts;
+	Assert(old_natts == numberOfAttributes);
+
+	d = SysCacheGetAttr(INDEXRELID, tuple, Anum_pg_index_indcollation, &isnull);
+	Assert(!isnull);
+	old_indcollation = (oidvector *) DatumGetPointer(d);
+
+	d = SysCacheGetAttr(INDEXRELID, tuple, Anum_pg_index_indclass, &isnull);
+	Assert(!isnull);
+	old_indclass = (oidvector *) DatumGetPointer(d);
+
+	family_am = accessMethodId == BTREE_AM_OID || accessMethodId == HASH_AM_OID;
+
+	for (i = 0; i < old_natts; i++)
+	{
+		Oid			old_class = old_indclass->values[i];
+		Oid			new_class = classObjectId[i];
+
+		if (!(old_indcollation->values[i] == collationObjectId[i]
+			  && (old_class == new_class
+				  || (family_am && (get_opclass_family(old_class)
+									== get_opclass_family(new_class))))))
+		{
+			ret = false;
+			break;
+		}
+	}
+
+	ReleaseSysCache(tuple);
+
+	/*
+	 * For btree and hash, exclusion operators need only fall in the same
+	 * operator family; ComputeIndexAttrs already verified that much.  If we
+	 * get this far, we know that the index operator family has not changed,
+	 * and we're done.  For other access methods, require exact matches for
+	 * all exclusion operators.
+	 */
+	if (ret && !family_am && indexInfo->ii_ExclusionOps != NULL)
+	{
+		Relation	irel;
+		Oid		   *old_operators, *old_procs;
+		uint16	   *old_strats;
+
+		/* Caller probably already holds a stronger lock. */
+		irel = index_open(oldId, AccessShareLock);
+		RelationGetExclusionInfo(irel, &old_operators, &old_procs, &old_strats);
+
+		for (i = 0; i < old_natts; i++)
+			if (old_operators[i] != indexInfo->ii_ExclusionOps[i])
+			{
+				ret = false;
+				break;
+			}
+
+		index_close(irel, NoLock);
+	}
+
+	return ret;
+}
+
+/*
  * DefineIndex
  *		Creates a new index.
  *
@@ -81,6 +273,8 @@ static char *ChooseIndexNameAddition(List *colnames);
  *		that a nonconflicting default name should be picked.
  * 'indexRelationId': normally InvalidOid, but during bootstrap can be
  *		nonzero to specify a preselected OID for the index.
+ * 'relFileNode': normally InvalidOid, but can be nonzero to specify existing
+ *		storage constituting a valid build of this index.
  * 'accessMethodName': name of the AM to use.
  * 'tableSpaceName': name of the tablespace to create the index in.
  *		NULL specifies using the appropriate default.
@@ -103,11 +297,14 @@ static char *ChooseIndexNameAddition(List *colnames);
  *		it will be filled later.
  * 'quiet': suppress the NOTICE chatter ordinarily provided for constraints.
  * 'concurrent': avoid blocking writers to the table while building.
+ *
+ * Returns the OID of the created index.
  */
-void
+Oid
 DefineIndex(RangeVar *heapRelation,
 			char *indexRelationName,
 			Oid indexRelationId,
+			Oid relFileNode,
 			char *accessMethodName,
 			char *tableSpaceName,
 			List *attributeList,
@@ -403,11 +600,17 @@ DefineIndex(RangeVar *heapRelation,
 	}
 
 	/*
+	 * A valid relFileNode implies that we already have a built form of the
+	 * index.  The caller should also decline any index build.
+	 */
+	Assert(!OidIsValid(relFileNode) || (skip_build && !concurrent));
+
+	/*
 	 * Make the catalog entries for the index, including constraints. Then, if
 	 * not skip_build || concurrent, actually build the index.
 	 */
 	indexRelationId =
-		index_create(rel, indexRelationName, indexRelationId,
+		index_create(rel, indexRelationName, indexRelationId, relFileNode,
 					 indexInfo, indexColNames,
 					 accessMethodId, tablespaceId,
 					 collationObjectId, classObjectId,
@@ -421,7 +624,7 @@ DefineIndex(RangeVar *heapRelation,
 	{
 		/* Close the heap and we're done, in the non-concurrent case */
 		heap_close(rel, NoLock);
-		return;
+		return indexRelationId;
 	}
 
 	/* save lockrelid and locktag for below, then close rel */
@@ -709,6 +912,8 @@ DefineIndex(RangeVar *heapRelation,
 	 * Last thing to do is release the session-level lock on the parent table.
 	 */
 	UnlockRelationIdForSession(&heaprelid, ShareUpdateExclusiveLock);
+
+	return indexRelationId;
 }
 
 
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 295a1ff6e63..82bb756c75e 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -347,7 +347,9 @@ static bool ATColumnChangeRequiresRewrite(Node *expr, AttrNumber varattno);
 static void ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 					  AlterTableCmd *cmd, LOCKMODE lockmode);
 static void ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab, LOCKMODE lockmode);
-static void ATPostAlterTypeParse(char *cmd, List **wqueue, LOCKMODE lockmode);
+static void ATPostAlterTypeParse(Oid oldId, char *cmd,
+					 List **wqueue, LOCKMODE lockmode, bool rewrite);
+static void TryReuseIndex(Oid oldId, IndexStmt *stmt);
 static void change_owner_recurse_to_sequences(Oid relationOid,
 								  Oid newOwnerId, LOCKMODE lockmode);
 static void ATExecClusterOn(Relation rel, const char *indexName, LOCKMODE lockmode);
@@ -5232,37 +5234,52 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
 	bool		check_rights;
 	bool		skip_build;
 	bool		quiet;
+	Oid			new_index;
 
 	Assert(IsA(stmt, IndexStmt));
 
 	/* suppress schema rights check when rebuilding existing index */
 	check_rights = !is_rebuild;
-	/* skip index build if phase 3 will have to rewrite table anyway */
-	skip_build = tab->rewrite;
+	/* skip index build if phase 3 will do it or we're reusing an old one */
+	skip_build = tab->rewrite || OidIsValid(stmt->oldNode);
 	/* suppress notices when rebuilding existing index */
 	quiet = is_rebuild;
 
 	/* The IndexStmt has already been through transformIndexStmt */
 
-	DefineIndex(stmt->relation, /* relation */
-				stmt->idxname,	/* index name */
-				InvalidOid,		/* no predefined OID */
-				stmt->accessMethod,		/* am name */
-				stmt->tableSpace,
-				stmt->indexParams,		/* parameters */
-				(Expr *) stmt->whereClause,
-				stmt->options,
-				stmt->excludeOpNames,
-				stmt->unique,
-				stmt->primary,
-				stmt->isconstraint,
-				stmt->deferrable,
-				stmt->initdeferred,
-				true,			/* is_alter_table */
-				check_rights,
-				skip_build,
-				quiet,
-				false);
+	new_index = DefineIndex(stmt->relation,		/* relation */
+							stmt->idxname,		/* index name */
+							InvalidOid, /* no predefined OID */
+							stmt->oldNode,
+							stmt->accessMethod, /* am name */
+							stmt->tableSpace,
+							stmt->indexParams,	/* parameters */
+							(Expr *) stmt->whereClause,
+							stmt->options,
+							stmt->excludeOpNames,
+							stmt->unique,
+							stmt->primary,
+							stmt->isconstraint,
+							stmt->deferrable,
+							stmt->initdeferred,
+							true,		/* is_alter_table */
+							check_rights,
+							skip_build,
+							quiet,
+							false);
+
+	/*
+	 * If TryReuseIndex() stashed a relfilenode for us, we used it for the new
+	 * index instead of building from scratch.  The DROP of the old edition of
+	 * this index will have scheduled the storage for deletion at commit, so
+	 * cancel that pending deletion.
+	 */
+	if (OidIsValid(stmt->oldNode))
+	{
+		Relation	irel = index_open(new_index, NoLock);
+		RelationPreserveStorage(irel->rd_node, true);
+		index_close(irel, NoLock);
+	}
 }
 
 /*
@@ -7389,7 +7406,8 @@ static void
 ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab, LOCKMODE lockmode)
 {
 	ObjectAddress obj;
-	ListCell   *l;
+	ListCell   *def_item;
+	ListCell   *oid_item;
 
 	/*
 	 * Re-parse the index and constraint definitions, and attach them to the
@@ -7399,10 +7417,14 @@ ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab, LOCKMODE lockmode)
 	 * that before dropping.  It's safe because the parser won't actually look
 	 * at the catalogs to detect the existing entry.
 	 */
-	foreach(l, tab->changedIndexDefs)
-		ATPostAlterTypeParse((char *) lfirst(l), wqueue, lockmode);
-	foreach(l, tab->changedConstraintDefs)
-		ATPostAlterTypeParse((char *) lfirst(l), wqueue, lockmode);
+	forboth(oid_item, tab->changedConstraintOids,
+			def_item, tab->changedConstraintDefs)
+		ATPostAlterTypeParse(lfirst_oid(oid_item), (char *) lfirst(def_item),
+							 wqueue, lockmode, tab->rewrite);
+	forboth(oid_item, tab->changedIndexOids,
+			def_item, tab->changedIndexDefs)
+		ATPostAlterTypeParse(lfirst_oid(oid_item), (char *) lfirst(def_item),
+							 wqueue, lockmode, tab->rewrite);
 
 	/*
 	 * Now we can drop the existing constraints and indexes --- constraints
@@ -7412,18 +7434,18 @@ ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab, LOCKMODE lockmode)
 	 * should be okay to use DROP_RESTRICT here, since nothing else should be
 	 * depending on these objects.
 	 */
-	foreach(l, tab->changedConstraintOids)
+	foreach(oid_item, tab->changedConstraintOids)
 	{
 		obj.classId = ConstraintRelationId;
-		obj.objectId = lfirst_oid(l);
+		obj.objectId = lfirst_oid(oid_item);
 		obj.objectSubId = 0;
 		performDeletion(&obj, DROP_RESTRICT);
 	}
 
-	foreach(l, tab->changedIndexOids)
+	foreach(oid_item, tab->changedIndexOids)
 	{
 		obj.classId = RelationRelationId;
-		obj.objectId = lfirst_oid(l);
+		obj.objectId = lfirst_oid(oid_item);
 		obj.objectSubId = 0;
 		performDeletion(&obj, DROP_RESTRICT);
 	}
@@ -7435,7 +7457,8 @@ ATPostAlterTypeCleanup(List **wqueue, AlteredTableInfo *tab, LOCKMODE lockmode)
 }
 
 static void
-ATPostAlterTypeParse(char *cmd, List **wqueue, LOCKMODE lockmode)
+ATPostAlterTypeParse(Oid oldId, char *cmd,
+					 List **wqueue, LOCKMODE lockmode, bool rewrite)
 {
 	List	   *raw_parsetree_list;
 	List	   *querytree_list;
@@ -7482,6 +7505,9 @@ ATPostAlterTypeParse(char *cmd, List **wqueue, LOCKMODE lockmode)
 					IndexStmt  *stmt = (IndexStmt *) stm;
 					AlterTableCmd *newcmd;
 
+					if (!rewrite)
+						TryReuseIndex(oldId, stmt);
+
 					rel = relation_openrv(stmt->relation, lockmode);
 					tab = ATGetQueueEntry(wqueue, rel);
 					newcmd = makeNode(AlterTableCmd);
@@ -7506,6 +7532,10 @@ ATPostAlterTypeParse(char *cmd, List **wqueue, LOCKMODE lockmode)
 						switch (cmd->subtype)
 						{
 							case AT_AddIndex:
+								Assert(IsA(cmd->def, IndexStmt));
+								if (!rewrite)
+									TryReuseIndex(get_constraint_index(oldId),
+												  (IndexStmt *) cmd->def);
 								cmd->subtype = AT_ReAddIndex;
 								tab->subcmds[AT_PASS_OLD_INDEX] =
 									lappend(tab->subcmds[AT_PASS_OLD_INDEX], cmd);
@@ -7529,6 +7559,26 @@ ATPostAlterTypeParse(char *cmd, List **wqueue, LOCKMODE lockmode)
 	}
 }
 
+/*
+ * Subroutine for ATPostAlterTypeParse().  Calls out to CheckIndexCompatible()
+ * for the real analysis, then mutates the IndexStmt based on that verdict.
+*/
+static void
+TryReuseIndex(Oid oldId, IndexStmt *stmt)
+{
+
+	if (CheckIndexCompatible(oldId,
+							 stmt->relation,
+							 stmt->accessMethod,
+							 stmt->indexParams,
+							 stmt->excludeOpNames))
+	{
+		Relation irel = index_open(oldId, NoLock);
+		stmt->oldNode = irel->rd_node.relNode;
+		index_close(irel, NoLock);
+	}
+}
+
 
 /*
  * ALTER TABLE OWNER
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index c9133ddfa19..7a5145621f3 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2807,6 +2807,7 @@ _copyIndexStmt(IndexStmt *from)
 	COPY_NODE_FIELD(whereClause);
 	COPY_NODE_FIELD(excludeOpNames);
 	COPY_SCALAR_FIELD(indexOid);
+	COPY_SCALAR_FIELD(oldNode);
 	COPY_SCALAR_FIELD(unique);
 	COPY_SCALAR_FIELD(primary);
 	COPY_SCALAR_FIELD(isconstraint);
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 3a0267c4191..4052a9a1fc9 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -1245,6 +1245,7 @@ _equalIndexStmt(IndexStmt *a, IndexStmt *b)
 	COMPARE_NODE_FIELD(whereClause);
 	COMPARE_NODE_FIELD(excludeOpNames);
 	COMPARE_SCALAR_FIELD(indexOid);
+	COMPARE_SCALAR_FIELD(oldNode);
 	COMPARE_SCALAR_FIELD(unique);
 	COMPARE_SCALAR_FIELD(primary);
 	COMPARE_SCALAR_FIELD(isconstraint);
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index 681f5f85bc7..b5be09af1a0 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -1976,6 +1976,7 @@ _outIndexStmt(StringInfo str, IndexStmt *node)
 	WRITE_NODE_FIELD(whereClause);
 	WRITE_NODE_FIELD(excludeOpNames);
 	WRITE_OID_FIELD(indexOid);
+	WRITE_OID_FIELD(oldNode);
 	WRITE_BOOL_FIELD(unique);
 	WRITE_BOOL_FIELD(primary);
 	WRITE_BOOL_FIELD(isconstraint);
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 0698cafbce1..0749227398b 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -953,6 +953,7 @@ standard_ProcessUtility(Node *parsetree,
 				DefineIndex(stmt->relation,		/* relation */
 							stmt->idxname,		/* index name */
 							InvalidOid, /* no predefined OID */
+							InvalidOid, /* no previous storage */
 							stmt->accessMethod, /* am name */
 							stmt->tableSpace,
 							stmt->indexParams,	/* parameters */
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 0b9d77a61d2..809222bfe95 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -2395,6 +2395,7 @@ RelationBuildLocalRelation(const char *relname,
 						   Oid relnamespace,
 						   TupleDesc tupDesc,
 						   Oid relid,
+						   Oid relfilenode,
 						   Oid reltablespace,
 						   bool shared_relation,
 						   bool mapped_relation,
@@ -2529,10 +2530,8 @@ RelationBuildLocalRelation(const char *relname,
 
 	/*
 	 * Insert relation physical and logical identifiers (OIDs) into the right
-	 * places.	Note that the physical ID (relfilenode) is initially the same
-	 * as the logical ID (OID); except that for a mapped relation, we set
-	 * relfilenode to zero and rely on RelationInitPhysicalAddr to consult the
-	 * map.
+	 * places.  For a mapped relation, we set relfilenode to zero and rely on
+	 * RelationInitPhysicalAddr to consult the map.
 	 */
 	rel->rd_rel->relisshared = shared_relation;
 
@@ -2547,10 +2546,10 @@ RelationBuildLocalRelation(const char *relname,
 	{
 		rel->rd_rel->relfilenode = InvalidOid;
 		/* Add it to the active mapping information */
-		RelationMapUpdateMap(relid, relid, shared_relation, true);
+		RelationMapUpdateMap(relid, relfilenode, shared_relation, true);
 	}
 	else
-		rel->rd_rel->relfilenode = relid;
+		rel->rd_rel->relfilenode = relfilenode;
 
 	RelationInitLockInfo(rel);	/* see lmgr.c */
 
diff --git a/src/backend/utils/cache/relmapper.c b/src/backend/utils/cache/relmapper.c
index a19ee28b53b..b04dc9e3497 100644
--- a/src/backend/utils/cache/relmapper.c
+++ b/src/backend/utils/cache/relmapper.c
@@ -792,7 +792,7 @@ write_relmap_file(bool shared, RelMapFile *newmap,
 			rnode.spcNode = tsid;
 			rnode.dbNode = dbid;
 			rnode.relNode = newmap->mappings[i].mapfilenode;
-			RelationPreserveStorage(rnode);
+			RelationPreserveStorage(rnode, false);
 		}
 	}
 
diff --git a/src/include/catalog/heap.h b/src/include/catalog/heap.h
index 0b7b190dd10..aee2d88ebe2 100644
--- a/src/include/catalog/heap.h
+++ b/src/include/catalog/heap.h
@@ -39,6 +39,7 @@ extern Relation heap_create(const char *relname,
 			Oid relnamespace,
 			Oid reltablespace,
 			Oid relid,
+			Oid relfilenode,
 			TupleDesc tupDesc,
 			char relkind,
 			char relpersistence,
diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h
index 071db7f4012..8b78b05a2e6 100644
--- a/src/include/catalog/index.h
+++ b/src/include/catalog/index.h
@@ -35,6 +35,7 @@ extern void index_check_primary_key(Relation heapRel,
 extern Oid index_create(Relation heapRelation,
 			 const char *indexRelationName,
 			 Oid indexRelationId,
+			 Oid relFileNode,
 			 IndexInfo *indexInfo,
 			 List *indexColNames,
 			 Oid accessMethodObjectId,
diff --git a/src/include/catalog/storage.h b/src/include/catalog/storage.h
index 8dee8cf2251..6907d836708 100644
--- a/src/include/catalog/storage.h
+++ b/src/include/catalog/storage.h
@@ -22,7 +22,7 @@
 
 extern void RelationCreateStorage(RelFileNode rnode, char relpersistence);
 extern void RelationDropStorage(Relation rel);
-extern void RelationPreserveStorage(RelFileNode rnode);
+extern void RelationPreserveStorage(RelFileNode rnode, bool atCommit);
 extern void RelationTruncate(Relation rel, BlockNumber nblocks);
 
 /*
diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h
index bbc024f50cc..81c515ebed3 100644
--- a/src/include/commands/defrem.h
+++ b/src/include/commands/defrem.h
@@ -18,9 +18,10 @@
 
 
 /* commands/indexcmds.c */
-extern void DefineIndex(RangeVar *heapRelation,
+extern Oid DefineIndex(RangeVar *heapRelation,
 			char *indexRelationName,
 			Oid indexRelationId,
+			Oid relFileNode,
 			char *accessMethodName,
 			char *tableSpaceName,
 			List *attributeList,
@@ -49,6 +50,11 @@ extern char *ChooseIndexName(const char *tabname, Oid namespaceId,
 				List *colnames, List *exclusionOpNames,
 				bool primary, bool isconstraint);
 extern List *ChooseIndexColumnNames(List *indexElems);
+extern bool CheckIndexCompatible(Oid oldId,
+					 RangeVar *heapRelation,
+					 char *accessMethodName,
+					 List *attributeList,
+					 List *exclusionOpNames);
 extern Oid	GetDefaultOpClass(Oid type_id, Oid am_id);
 
 /* commands/functioncmds.c */
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 00c1269bd14..92e40d3fb58 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2062,6 +2062,7 @@ typedef struct IndexStmt
 	Node	   *whereClause;	/* qualification (partial-index predicate) */
 	List	   *excludeOpNames; /* exclusion operator names, or NIL if none */
 	Oid			indexOid;		/* OID of an existing index, if any */
+	Oid			oldNode;		/* relfilenode of my former self */
 	bool		unique;			/* is index unique? */
 	bool		primary;		/* is index on primary key? */
 	bool		isconstraint;	/* is it from a CONSTRAINT clause? */
diff --git a/src/include/utils/relcache.h b/src/include/utils/relcache.h
index 1f4def56845..ef1692cfa84 100644
--- a/src/include/utils/relcache.h
+++ b/src/include/utils/relcache.h
@@ -67,6 +67,7 @@ extern Relation RelationBuildLocalRelation(const char *relname,
 						   Oid relnamespace,
 						   TupleDesc tupDesc,
 						   Oid relid,
+						   Oid relfilenode,
 						   Oid reltablespace,
 						   bool shared_relation,
 						   bool mapped_relation,