database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-06-17 17:02:08 +03:00
Code Activity

Allow UNIQUE indexes on partitioned tables

Browse Source 
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
This commit is contained in:
Alvaro Herrera
2018-02-19 16:59:37 -03:00
parent 524d64ea8e
commit eb7ed3f306
27 changed files with 907 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
src/backend/commands/tablecmds.c
View File

@ -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();