From db7c60ed2647d566be9057d161a95ef8d2270dad Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Wed, 18 Oct 2017 19:49:15 +0200
Subject: [PATCH v2 4/7] Match/create indexes during ATTACH PARTITION

---
 src/backend/catalog/index.c      | 45 ++++++++++++++++++-
 src/backend/commands/indexcmds.c | 25 +++++++++++
 src/backend/commands/tablecmds.c | 97 ++++++++++++++++++++++++++++++++++++++++
 src/include/catalog/index.h      |  4 ++
 4 files changed, 170 insertions(+), 1 deletion(-)

diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 8d15db0ec1..8588b470b7 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -99,6 +99,7 @@ static void InitializeAttributeOids(Relation indexRelation,
 						int numatts, Oid indexoid);
 static void AppendAttributeTuples(Relation indexRelation, int numatts);
 static void UpdateIndexRelation(Oid indexoid, Oid heapoid,
+					Oid parentIndexId,
 					IndexInfo *indexInfo,
 					Oid *collationOids,
 					Oid *classOids,
@@ -552,6 +553,7 @@ AppendAttributeTuples(Relation indexRelation, int numatts)
 static void
 UpdateIndexRelation(Oid indexoid,
 					Oid heapoid,
+					Oid parentIndexOid,
 					IndexInfo *indexInfo,
 					Oid *collationOids,
 					Oid *classOids,
@@ -625,6 +627,7 @@ UpdateIndexRelation(Oid indexoid,
 
 	values[Anum_pg_index_indexrelid - 1] = ObjectIdGetDatum(indexoid);
 	values[Anum_pg_index_indrelid - 1] = ObjectIdGetDatum(heapoid);
+	values[Anum_pg_index_indparentidx - 1 ] = ObjectIdGetDatum(parentIndexOid);
 	values[Anum_pg_index_indnatts - 1] = Int16GetDatum(indexInfo->ii_NumIndexAttrs);
 	values[Anum_pg_index_indisunique - 1] = BoolGetDatum(indexInfo->ii_Unique);
 	values[Anum_pg_index_indisprimary - 1] = BoolGetDatum(primary);
@@ -932,7 +935,8 @@ index_create(Relation heapRelation,
 	 *	  (Or, could define a rule to maintain the predicate) --Nels, Feb '92
 	 * ----------------
 	 */
-	UpdateIndexRelation(indexRelationId, heapRelationId, indexInfo,
+	UpdateIndexRelation(indexRelationId, heapRelationId, parentIndexRelid,
+						indexInfo,
 						collationObjectId, classObjectId, coloptions,
 						isprimary, is_exclusion,
 						(constr_flags & INDEX_CONSTR_CREATE_DEFERRABLE) == 0,
@@ -1719,6 +1723,45 @@ BuildIndexInfo(Relation index)
 	return ii;
 }
 
+/*
+ * Compare two IndexInfos, and return true if they are similar enough that an
+ * index built with one can pass as an index built with the other.  If an
+ * attmap is given, the indexes are from tables where the columns might be in
+ * different physical locations, so use the map to match the column by name.
+ */
+bool
+CompareIndexInfo(IndexInfo *info1, IndexInfo *info2, AttrNumber *attmap)
+{
+	int		i;
+
+	if (info1->ii_NumIndexAttrs != info2->ii_NumIndexAttrs)
+		return false;
+
+	for (i = 0; i < info1->ii_NumIndexAttrs; i++)
+	{
+		/* XXX use attmap here */
+		if (info1->ii_KeyAttrNumbers[i] != info2->ii_KeyAttrNumbers[i])
+			return false;
+	}
+
+	/* Expression indexes are currently not considered equal.  Someday ... */
+	if (info1->ii_Expressions != NIL || info2->ii_Expressions != NIL)
+		return false;
+
+	/* Can this be relaxed? */
+	if (!equal(info1->ii_Predicate, info2->ii_Predicate))
+		return false;
+
+	/* Probably this can be relaxed someday */
+	if (info1->ii_ExclusionOps != NULL || info2->ii_ExclusionOps != NULL)
+		return false;
+
+	if (info1->ii_Unique != info2->ii_Unique)
+		return false;
+
+	return true;
+}
+
 /* ----------------
  *		BuildSpeculativeIndexInfo
  *			Add extra state to IndexInfo record
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index a9a8969422..b22ad68c48 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -2181,3 +2181,28 @@ ReindexPartitionedIndex(Relation parentIdx)
 
 	heap_close(parentIdx, AccessExclusiveLock);
 }
+
+/*
+ * Update the pg_index tuple corresponding to the given index on a partition
+ * to indicate that the given index OID is now its parent partitioned index.
+ */
+void
+indexSetParentIndex(Relation idx, Oid parentOid)
+{
+	Relation	pgindex;
+	HeapTuple	indTup;
+	Form_pg_index indForm;
+
+	/* Make sure this is an index */
+	Assert(idx->rd_rel->relkind == RELKIND_INDEX ||
+		   idx->rd_rel->relkind == RELKIND_PARTITIONED_INDEX);
+
+	pgindex = heap_open(IndexRelationId, RowExclusiveLock);
+	indTup = idx->rd_indextuple;
+	indForm = (Form_pg_index) GETSTRUCT(indTup);
+	indForm->indparentidx = parentOid;
+
+	CatalogTupleUpdate(pgindex, &(indTup->t_self), indTup);
+
+	heap_close(pgindex, RowExclusiveLock);
+}
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 17cb2f05e7..6ffe98d10f 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -13953,6 +13953,103 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
 	StorePartitionBound(attachrel, rel, cmd->bound);
 
 	/*
+	 * Ensure a correct set of indexes in the partition.
+	 * XXX probably this should be a new function.
+	 */
+	{
+		AttrNumber *attmap = NULL;
+		List	   *idxes;
+		List	   *attachRelIdxs;
+		Relation   *attachrelIdxRels;
+		IndexInfo **attachInfos;
+		int			i;
+		ListCell   *cell;
+
+		idxes = RelationGetIndexList(rel);
+		attachRelIdxs = RelationGetIndexList(attachrel);
+		attachrelIdxRels = palloc(sizeof(Relation) * list_length(attachRelIdxs));
+		attachInfos = palloc(sizeof(IndexInfo *) * list_length(attachRelIdxs));
+
+		i = 0;
+		foreach(cell, attachRelIdxs)
+		{
+			Oid			cldIdxId = lfirst_oid(cell);
+
+			attachrelIdxRels[i] = index_open(cldIdxId, AccessShareLock);
+			attachInfos[i] = BuildIndexInfo(attachrelIdxRels[i]);
+			i++;
+		}
+
+		/*
+		 * For each index on the partitioned table, find a match in the table
+		 * being attached as partition; if one is not found, create one.
+		 */
+		foreach(cell, idxes)
+		{
+			Oid			idx = lfirst_oid(cell);
+			Relation	idxRel = index_open(idx, AccessShareLock);
+			IndexInfo  *info;
+			bool		found = false;
+
+			if (idxRel->rd_rel->relkind != RELKIND_PARTITIONED_INDEX)
+			{
+				index_close(idxRel, AccessShareLock);
+				continue;
+			}
+			info = BuildIndexInfo(idxRel);
+			if (attmap == NULL)
+				attmap =
+					convert_tuples_by_name_map(RelationGetDescr(attachrel),
+											   RelationGetDescr(rel),
+											   gettext_noop("could not convert row type"));
+
+			for (i = 0; i < list_length(attachRelIdxs); i++)
+			{
+				if (CompareIndexInfo(info, attachInfos[i], attmap))
+				{
+					/* bingo. */
+					indexSetParentIndex(attachrelIdxRels[i], idx);
+					found = true;
+				}
+			}
+			if (!found)
+			{
+				RangeVar   *heap_rv;
+				IndexStmt  *stmt;
+
+				heap_rv = makeRangeVar(get_namespace_name(RelationGetNamespace(attachrel)),
+									   RelationGetRelationName(attachrel),
+									   -1);
+				stmt = generateClonedIndexStmt(heap_rv, idxRel, attmap,
+											   RelationGetDescr(rel)->natts);
+				/* XXX emit a DDL event here */
+				DefineIndex(RelationGetRelid(attachrel),
+							stmt,
+							InvalidOid,
+							RelationGetRelid(idxRel),
+							false, false, false, false, false);
+			}
+
+			index_close(idxRel, AccessShareLock);
+		}
+
+		/* Clean up. */
+		if (attmap)
+			pfree(attmap);
+
+		for (i = 0; i < list_length(attachRelIdxs); i++)
+		{
+			pfree(attachInfos[i]);
+			index_close(attachrelIdxRels[i], AccessShareLock);
+		}
+
+		if (idxes)
+			pfree(idxes);
+		if (attachRelIdxs)
+			pfree(attachRelIdxs);
+	}
+
+	/*
 	 * Generate partition constraint from the partition bound specification.
 	 * If the parent itself is a partition, make sure to include its
 	 * constraint as well.
diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h
index b9914c5d15..8438ffe4ce 100644
--- a/src/include/catalog/index.h
+++ b/src/include/catalog/index.h
@@ -86,6 +86,8 @@ extern void index_drop(Oid indexId, bool concurrent);
 
 extern IndexInfo *BuildIndexInfo(Relation index);
 
+extern bool CompareIndexInfo(IndexInfo *info1, IndexInfo *info2, AttrNumber *attmap);
+
 extern void BuildSpeculativeIndexInfo(Relation index, IndexInfo *ii);
 
 extern void FormIndexDatum(IndexInfo *indexInfo,
@@ -136,4 +138,6 @@ extern bool ReindexIsProcessingHeap(Oid heapOid);
 extern bool ReindexIsProcessingIndex(Oid indexOid);
 extern Oid	IndexGetRelation(Oid indexId, bool missing_ok);
 
+extern void indexSetParentIndex(Relation idx, Oid parentOid);
+
 #endif							/* INDEX_H */
-- 
2.11.0

