database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-06-14 18:42:34 +03:00
Code Activity

Code beautification for ATExecAttachPartition.

Browse Source 
Amit Langote

Discussion: http://postgr.es/m/CAFjFpReT_kq_uwU_B8aWDxR7jNGE=P0iELycdq5oupi=xSQTOw@mail.gmail.com
This commit is contained in:
Robert Haas
2017-08-03 14:16:33 -04:00
parent 86705aa8c3
commit 583df3b5c5
1 changed files with 61 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -13419,10 +13419,10 @@ ComputePartitionAttrs(Relation rel, List *partParams, AttrNumber *partattrs,
static ObjectAddress static ObjectAddress
ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd) ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
{ {
Relation attachRel, Relation attachrel,
catalog; catalog;
List *childrels; List *attachrel_children;
TupleConstr *attachRel_constr; TupleConstr *attachrel_constr;
List *partConstraint, List *partConstraint,
*existConstraint; *existConstraint;
SysScanDesc scan; SysScanDesc scan;
@ -13434,22 +13434,22 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
ObjectAddress address; ObjectAddress address;
const char *trigger_name; const char *trigger_name;
attachRel = heap_openrv(cmd->name, AccessExclusiveLock); attachrel = heap_openrv(cmd->name, AccessExclusiveLock);
/* /*
* Must be owner of both parent and source table -- parent was checked by * Must be owner of both parent and source table -- parent was checked by
* ATSimplePermissions call in ATPrepCmd * ATSimplePermissions call in ATPrepCmd
*/ */
ATSimplePermissions(attachRel, ATT_TABLE | ATT_FOREIGN_TABLE); ATSimplePermissions(attachrel, ATT_TABLE | ATT_FOREIGN_TABLE);
/* A partition can only have one parent */ /* A partition can only have one parent */
if (attachRel->rd_rel->relispartition) if (attachrel->rd_rel->relispartition)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE), (errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("\"%s\" is already a partition", errmsg("\"%s\" is already a partition",
RelationGetRelationName(attachRel)))); RelationGetRelationName(attachrel))));
if (OidIsValid(attachRel->rd_rel->reloftype)) if (OidIsValid(attachrel->rd_rel->reloftype))
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE), (errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot attach a typed table as partition"))); errmsg("cannot attach a typed table as partition")));
@ -13462,7 +13462,7 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
ScanKeyInit(&skey, ScanKeyInit(&skey,
Anum_pg_inherits_inhrelid, Anum_pg_inherits_inhrelid,
BTEqualStrategyNumber, F_OIDEQ, BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(RelationGetRelid(attachRel))); ObjectIdGetDatum(RelationGetRelid(attachrel)));
scan = systable_beginscan(catalog, InheritsRelidSeqnoIndexId, true, scan = systable_beginscan(catalog, InheritsRelidSeqnoIndexId, true,
NULL, 1, &skey); NULL, 1, &skey);
if (HeapTupleIsValid(systable_getnext(scan))) if (HeapTupleIsValid(systable_getnext(scan)))
@ -13475,11 +13475,11 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
ScanKeyInit(&skey, ScanKeyInit(&skey,
Anum_pg_inherits_inhparent, Anum_pg_inherits_inhparent,
BTEqualStrategyNumber, F_OIDEQ, BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(RelationGetRelid(attachRel))); ObjectIdGetDatum(RelationGetRelid(attachrel)));
scan = systable_beginscan(catalog, InheritsParentIndexId, true, NULL, scan = systable_beginscan(catalog, InheritsParentIndexId, true, NULL,
1, &skey); 1, &skey);
if (HeapTupleIsValid(systable_getnext(scan)) && if (HeapTupleIsValid(systable_getnext(scan)) &&
attachRel->rd_rel->relkind == RELKIND_RELATION) attachrel->rd_rel->relkind == RELKIND_RELATION)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE), (errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot attach inheritance parent as partition"))); errmsg("cannot attach inheritance parent as partition")));
@ -13487,22 +13487,22 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
heap_close(catalog, AccessShareLock); heap_close(catalog, AccessShareLock);
/* /*
* Prevent circularity by seeing if rel is a partition of attachRel. (In * Prevent circularity by seeing if rel is a partition of attachrel. (In
* particular, this disallows making a rel a partition of itself.) * particular, this disallows making a rel a partition of itself.)
*/ */
childrels = find_all_inheritors(RelationGetRelid(attachRel), attachrel_children = find_all_inheritors(RelationGetRelid(attachrel),
AccessShareLock, NULL); AccessShareLock, NULL);
if (list_member_oid(childrels, RelationGetRelid(rel))) if (list_member_oid(attachrel_children, RelationGetRelid(rel)))
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_DUPLICATE_TABLE), (errcode(ERRCODE_DUPLICATE_TABLE),
errmsg("circular inheritance not allowed"), errmsg("circular inheritance not allowed"),
errdetail("\"%s\" is already a child of \"%s\".", errdetail("\"%s\" is already a child of \"%s\".",
RelationGetRelationName(rel), RelationGetRelationName(rel),
RelationGetRelationName(attachRel)))); RelationGetRelationName(attachrel))));
/* Temp parent cannot have a partition that is itself not a temp */ /* Temp parent cannot have a partition that is itself not a temp */
if (rel->rd_rel->relpersistence == RELPERSISTENCE_TEMP && if (rel->rd_rel->relpersistence == RELPERSISTENCE_TEMP &&
attachRel->rd_rel->relpersistence != RELPERSISTENCE_TEMP) attachrel->rd_rel->relpersistence != RELPERSISTENCE_TEMP)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE), (errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot attach a permanent relation as partition of temporary relation \"%s\"", errmsg("cannot attach a permanent relation as partition of temporary relation \"%s\"",
@ -13516,30 +13516,30 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
errmsg("cannot attach as partition of temporary relation of another session"))); errmsg("cannot attach as partition of temporary relation of another session")));
/* Ditto for the partition */ /* Ditto for the partition */
if (attachRel->rd_rel->relpersistence == RELPERSISTENCE_TEMP && if (attachrel->rd_rel->relpersistence == RELPERSISTENCE_TEMP &&
!attachRel->rd_islocaltemp) !attachrel->rd_islocaltemp)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE), (errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot attach temporary relation of another session as partition"))); errmsg("cannot attach temporary relation of another session as partition")));
/* If parent has OIDs then child must have OIDs */ /* If parent has OIDs then child must have OIDs */
if (rel->rd_rel->relhasoids && !attachRel->rd_rel->relhasoids) if (rel->rd_rel->relhasoids && !attachrel->rd_rel->relhasoids)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE), (errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot attach table \"%s\" without OIDs as partition of" errmsg("cannot attach table \"%s\" without OIDs as partition of"
" table \"%s\" with OIDs", RelationGetRelationName(attachRel), " table \"%s\" with OIDs", RelationGetRelationName(attachrel),
RelationGetRelationName(rel)))); RelationGetRelationName(rel))));
/* OTOH, if parent doesn't have them, do not allow in attachRel either */ /* OTOH, if parent doesn't have them, do not allow in attachrel either */
if (attachRel->rd_rel->relhasoids && !rel->rd_rel->relhasoids) if (attachrel->rd_rel->relhasoids && !rel->rd_rel->relhasoids)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE), (errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot attach table \"%s\" with OIDs as partition of table" errmsg("cannot attach table \"%s\" with OIDs as partition of table"
" \"%s\" without OIDs", RelationGetRelationName(attachRel), " \"%s\" without OIDs", RelationGetRelationName(attachrel),
RelationGetRelationName(rel)))); RelationGetRelationName(rel))));
/* Check if there are any columns in attachRel that aren't in the parent */ /* Check if there are any columns in attachrel that aren't in the parent */
tupleDesc = RelationGetDescr(attachRel); tupleDesc = RelationGetDescr(attachrel);
natts = tupleDesc->natts; natts = tupleDesc->natts;
for (attno = 1; attno <= natts; attno++) for (attno = 1; attno <= natts; attno++)
{ {
@ -13557,7 +13557,7 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH), (errcode(ERRCODE_DATATYPE_MISMATCH),
errmsg("table \"%s\" contains column \"%s\" not found in parent \"%s\"", errmsg("table \"%s\" contains column \"%s\" not found in parent \"%s\"",
RelationGetRelationName(attachRel), attributeName, RelationGetRelationName(attachrel), attributeName,
RelationGetRelationName(rel)), RelationGetRelationName(rel)),
errdetail("New partition should contain only the columns present in parent."))); errdetail("New partition should contain only the columns present in parent.")));
} }
@ -13567,34 +13567,34 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
* currently don't allow it to become a partition. See also prohibitions * currently don't allow it to become a partition. See also prohibitions
* in ATExecAddInherit() and CreateTrigger(). * in ATExecAddInherit() and CreateTrigger().
*/ */
trigger_name = FindTriggerIncompatibleWithInheritance(attachRel->trigdesc); trigger_name = FindTriggerIncompatibleWithInheritance(attachrel->trigdesc);
if (trigger_name != NULL) if (trigger_name != NULL)
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED), (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("trigger \"%s\" prevents table \"%s\" from becoming a partition", errmsg("trigger \"%s\" prevents table \"%s\" from becoming a partition",
trigger_name, RelationGetRelationName(attachRel)), trigger_name, RelationGetRelationName(attachrel)),
errdetail("ROW triggers with transition tables are not supported on partitions"))); errdetail("ROW triggers with transition tables are not supported on partitions")));
/* OK to create inheritance. Rest of the checks performed there */ /* OK to create inheritance. Rest of the checks performed there */
CreateInheritance(attachRel, rel); CreateInheritance(attachrel, rel);
/* /*
* Check that the new partition's bound is valid and does not overlap any * Check that the new partition's bound is valid and does not overlap any
* of existing partitions of the parent - note that it does not return on * of existing partitions of the parent - note that it does not return on
* error. * error.
*/ */
check_new_partition_bound(RelationGetRelationName(attachRel), rel, check_new_partition_bound(RelationGetRelationName(attachrel), rel,
cmd->bound); cmd->bound);
/* Update the pg_class entry. */ /* Update the pg_class entry. */
StorePartitionBound(attachRel, rel, cmd->bound); StorePartitionBound(attachrel, rel, cmd->bound);
/* /*
* Generate partition constraint from the partition bound specification. * Generate partition constraint from the partition bound specification.
* If the parent itself is a partition, make sure to include its * If the parent itself is a partition, make sure to include its
* constraint as well. * constraint as well.
*/ */
partConstraint = list_concat(get_qual_from_partbound(attachRel, rel, partConstraint = list_concat(get_qual_from_partbound(attachrel, rel,
cmd->bound), cmd->bound),
RelationGetPartitionQual(rel)); RelationGetPartitionQual(rel));
partConstraint = (List *) eval_const_expressions(NULL, partConstraint = (List *) eval_const_expressions(NULL,
@ -13612,20 +13612,20 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
* There is a case in which we cannot rely on just the result of the * There is a case in which we cannot rely on just the result of the
* proof. * proof.
*/ */
attachRel_constr = tupleDesc->constr; attachrel_constr = tupleDesc->constr;
existConstraint = NIL; existConstraint = NIL;
if (attachRel_constr != NULL) if (attachrel_constr != NULL)
{ {
int num_check = attachRel_constr->num_check; int num_check = attachrel_constr->num_check;
int i; int i;
if (attachRel_constr->has_not_null) if (attachrel_constr->has_not_null)
{ {
int natts = attachRel->rd_att->natts; int natts = attachrel->rd_att->natts;
for (i = 1; i <= natts; i++) for (i = 1; i <= natts; i++)
{ {
Form_pg_attribute att = attachRel->rd_att->attrs[i - 1]; Form_pg_attribute att = attachrel->rd_att->attrs[i - 1];
if (att->attnotnull && !att->attisdropped) if (att->attnotnull && !att->attisdropped)
{ {
@ -13659,10 +13659,10 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
* If this constraint hasn't been fully validated yet, we must * If this constraint hasn't been fully validated yet, we must
* ignore it here. * ignore it here.
*/ */
if (!attachRel_constr->check[i].ccvalid) if (!attachrel_constr->check[i].ccvalid)
continue; continue;
cexpr = stringToNode(attachRel_constr->check[i].ccbin); cexpr = stringToNode(attachrel_constr->check[i].ccbin);
/* /*
* Run each expression through const-simplification and * Run each expression through const-simplification and
@ -13684,28 +13684,25 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
skip_validate = true; skip_validate = true;
} }
/* It's safe to skip the validation scan after all */
if (skip_validate) if (skip_validate)
{
/* No need to scan the table after all. */
ereport(INFO, ereport(INFO,
(errmsg("partition constraint for table \"%s\" is implied by existing constraints", (errmsg("partition constraint for table \"%s\" is implied by existing constraints",
RelationGetRelationName(attachRel)))); RelationGetRelationName(attachrel))));
}
/* else
* Set up to have the table be scanned to validate the partition
* constraint (see partConstraint above). If it's a partitioned table, we
* instead schedule its leaf partitions to be scanned.
*/
if (!skip_validate)
{ {
/* Constraints proved insufficient, so we need to scan the table. */
List *all_parts; List *all_parts;
ListCell *lc; ListCell *lc;
/* Take an exclusive lock on the partitions to be checked */ /* Take an exclusive lock on the partitions to be checked */
if (attachRel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) if (attachrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
all_parts = find_all_inheritors(RelationGetRelid(attachRel), all_parts = find_all_inheritors(RelationGetRelid(attachrel),
AccessExclusiveLock, NULL); AccessExclusiveLock, NULL);
else else
all_parts = list_make1_oid(RelationGetRelid(attachRel)); all_parts = list_make1_oid(RelationGetRelid(attachrel));
foreach(lc, all_parts) foreach(lc, all_parts)
{ {
@ -13716,23 +13713,23 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
bool found_whole_row; bool found_whole_row;
/* Lock already taken */ /* Lock already taken */
if (part_relid != RelationGetRelid(attachRel)) if (part_relid != RelationGetRelid(attachrel))
part_rel = heap_open(part_relid, NoLock); part_rel = heap_open(part_relid, NoLock);
else else
part_rel = attachRel; part_rel = attachrel;
/* /*
* Skip if it's a partitioned table. Only RELKIND_RELATION * Skip if the partition is itself a partitioned table. We can
* relations (ie, leaf partitions) need to be scanned. * only ever scan RELKIND_RELATION relations.
*/ */
if (part_rel != attachRel && if (part_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
part_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
{ {
heap_close(part_rel, NoLock); if (part_rel != attachrel)
heap_close(part_rel, NoLock);
continue; continue;
} }
/* Grab a work queue entry */ /* Grab a work queue entry. */
tab = ATGetQueueEntry(wqueue, part_rel); tab = ATGetQueueEntry(wqueue, part_rel);
/* Adjust constraint to match this partition */ /* Adjust constraint to match this partition */
@ -13746,15 +13743,15 @@ ATExecAttachPartition(List **wqueue, Relation rel, PartitionCmd *cmd)
elog(ERROR, "unexpected whole-row reference found in partition key"); elog(ERROR, "unexpected whole-row reference found in partition key");
/* keep our lock until commit */ /* keep our lock until commit */
if (part_rel != attachRel) if (part_rel != attachrel)
heap_close(part_rel, NoLock); heap_close(part_rel, NoLock);
} }
} }
ObjectAddressSet(address, RelationRelationId, RelationGetRelid(attachRel)); ObjectAddressSet(address, RelationRelationId, RelationGetRelid(attachrel));
/* keep our lock until commit */ /* keep our lock until commit */
heap_close(attachRel, NoLock); heap_close(attachrel, NoLock);
return address; return address;
} }