Allow UNIQUE indexes on partitioned tables

If we restrict unique constraints on partitioned tables so that they
must always include the partition key, then our standard approach to
unique indexes already works --- each unique key is forced to exist
within a single partition, so enforcing the unique restriction in each
index individually is enough to have it enforced globally.  Therefore we
can implement unique indexes on partitions by simply removing a few
restrictions (and adding others.)

Discussion: https://postgr.es/m/20171222212921.hi6hg6pem2w2t36z@alvherre.pgsql
Discussion: https://postgr.es/m/20171229230607.3iib6b62fn3uaf47@alvherre.pgsql
Reviewed-by: Simon Riggs, Jesper Pedersen, Peter Eisentraut, Jaime
	Casanova, Amit Langote

1 files changed, 69 insertions, 2 deletions

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 87539d6c0b2..db6c8ff00e9 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -939,17 +939,20 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 			Relation	idxRel = index_open(lfirst_oid(cell), AccessShareLock);
 			AttrNumber *attmap;
 			IndexStmt  *idxstmt;
+			Oid			constraintOid;
 
 			attmap = convert_tuples_by_name_map(RelationGetDescr(rel),
 												RelationGetDescr(parent),
 												gettext_noop("could not convert row type"));
 			idxstmt =
 				generateClonedIndexStmt(NULL, RelationGetRelid(rel), idxRel,
-										attmap, RelationGetDescr(rel)->natts);
+										attmap, RelationGetDescr(rel)->natts,
+										&constraintOid);
 			DefineIndex(RelationGetRelid(rel),
 						idxstmt,
 						InvalidOid,
 						RelationGetRelid(idxRel),
+						constraintOid,
 						false, false, false, false, false);
 
 			index_close(idxRel, AccessShareLock);
@@ -6824,6 +6827,7 @@ ATExecAddIndex(AlteredTableInfo *tab, Relation rel,
 						  stmt,
 						  InvalidOid,	/* no predefined OID */
 						  InvalidOid,	/* no parent index */
+						  InvalidOid,	/* no parent constraint */
 						  true, /* is_alter_table */
 						  check_rights,
 						  false,	/* check_not_in_use - we did it already */
@@ -6869,6 +6873,15 @@ ATExecAddIndexConstraint(AlteredTableInfo *tab, Relation rel,
 	Assert(OidIsValid(index_oid));
 	Assert(stmt->isconstraint);
 
+	/*
+	 * Doing this on partitioned tables is not a simple feature to implement,
+	 * so let's punt for now.
+	 */
+	if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("ALTER TABLE / ADD CONSTRAINT USING INDEX is not supported on partitioned tables")));
+
 	indexRel = index_open(index_oid, AccessShareLock);
 
 	indexName = pstrdup(RelationGetRelationName(indexRel));
@@ -6916,6 +6929,7 @@ ATExecAddIndexConstraint(AlteredTableInfo *tab, Relation rel,
 
 	address = index_constraint_create(rel,
 									  index_oid,
+									  InvalidOid,
 									  indexInfo,
 									  constraintName,
 									  constraintType,
@@ -14147,6 +14161,7 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 		IndexInfo  *info;
 		AttrNumber *attmap;
 		bool		found = false;
+		Oid			constraintOid;
 
 		/*
 		 * Ignore indexes in the partitioned table other than partitioned
@@ -14163,6 +14178,7 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 		attmap = convert_tuples_by_name_map(RelationGetDescr(attachrel),
 											RelationGetDescr(rel),
 											gettext_noop("could not convert row type"));
+		constraintOid = get_relation_idx_constraint_oid(RelationGetRelid(rel), idx);
 
 		/*
 		 * Scan the list of existing indexes in the partition-to-be, and mark
@@ -14171,6 +14187,8 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 		 */
 		for (i = 0; i < list_length(attachRelIdxs); i++)
 		{
+			Oid		cldConstrOid = InvalidOid;
+
 			/* does this index have a parent?  if so, can't use it */
 			if (has_superclass(RelationGetRelid(attachrelIdxRels[i])))
 				continue;
@@ -14183,8 +14201,26 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 								 attmap,
 								 RelationGetDescr(rel)->natts))
 			{
+				/*
+				 * If this index is being created in the parent because of a
+				 * constraint, then the child needs to have a constraint also,
+				 * so look for one.  If there is no such constraint, this
+				 * index is no good, so keep looking.
+				 */
+				if (OidIsValid(constraintOid))
+				{
+					cldConstrOid =
+						get_relation_idx_constraint_oid(RelationGetRelid(attachrel),
+														RelationGetRelid(attachrelIdxRels[i]));
+					/* no dice */
+					if (!OidIsValid(cldConstrOid))
+						continue;
+				}
+
 				/* bingo. */
 				IndexSetParentIndex(attachrelIdxRels[i], idx);
+				if (OidIsValid(constraintOid))
+					ConstraintSetParentConstraint(cldConstrOid, constraintOid);
 				found = true;
 				break;
 			}
@@ -14197,12 +14233,15 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 		if (!found)
 		{
 			IndexStmt  *stmt;
+			Oid			constraintOid;
 
 			stmt = generateClonedIndexStmt(NULL, RelationGetRelid(attachrel),
 										   idxRel, attmap,
-										   RelationGetDescr(rel)->natts);
+										   RelationGetDescr(rel)->natts,
+										   &constraintOid);
 			DefineIndex(RelationGetRelid(attachrel), stmt, InvalidOid,
 						RelationGetRelid(idxRel),
+						constraintOid,
 						false, false, false, false, false);
 		}
 
@@ -14445,6 +14484,8 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name)
 		bool		found;
 		int			i;
 		PartitionDesc partDesc;
+		Oid			constraintOid,
+					cldConstrId = InvalidOid;
 
 		/*
 		 * If this partition already has an index attached, refuse the operation.
@@ -14500,8 +14541,34 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name)
 							RelationGetRelationName(parentIdx)),
 					 errdetail("The index definitions do not match.")));
 
+		/*
+		 * If there is a constraint in the parent, make sure there is one
+		 * in the child too.
+		 */
+		constraintOid = get_relation_idx_constraint_oid(RelationGetRelid(parentTbl),
+														RelationGetRelid(parentIdx));
+
+		if (OidIsValid(constraintOid))
+		{
+			cldConstrId = get_relation_idx_constraint_oid(RelationGetRelid(partTbl),
+														  partIdxId);
+			if (!OidIsValid(cldConstrId))
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+						 errmsg("cannot attach index \"%s\" as a partition of index \"%s\"",
+								RelationGetRelationName(partIdx),
+								RelationGetRelationName(parentIdx)),
+						 errdetail("The index \"%s\" belongs to a constraint in table \"%s\" but no constraint exists for index \"%s\".",
+								RelationGetRelationName(parentIdx),
+								RelationGetRelationName(parentTbl),
+								RelationGetRelationName(partIdx))));
+		}
+
 		/* All good -- do it */
 		IndexSetParentIndex(partIdx, RelationGetRelid(parentIdx));
+		if (OidIsValid(constraintOid))
+			ConstraintSetParentConstraint(cldConstrId, constraintOid);
+
 		pfree(attmap);
 
 		CommandCounterIncrement();