diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c index 2063abb8ae..abc704d3e7 100644 --- a/src/backend/catalog/pg_constraint.c +++ b/src/backend/catalog/pg_constraint.c @@ -403,6 +403,7 @@ CloneForeignKeyConstraints(Oid parentId, Oid relationId, List **cloned) TupleDesc tupdesc; HeapTuple tuple; AttrNumber *attmap; + List *partFKs; parentRel = heap_open(parentId, NoLock); /* already got lock */ /* see ATAddForeignKeyConstraint about lock level */ @@ -425,6 +426,8 @@ CloneForeignKeyConstraints(Oid parentId, Oid relationId, List **cloned) scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId, true, NULL, 1, &key); + partFKs = copyObject(RelationGetFKeyList(rel)); + while ((tuple = systable_getnext(scan)) != NULL) { Form_pg_constraint constrForm = (Form_pg_constraint) GETSTRUCT(tuple); @@ -435,11 +438,13 @@ CloneForeignKeyConstraints(Oid parentId, Oid relationId, List **cloned) Oid conppeqop[INDEX_MAX_KEYS]; Oid conffeqop[INDEX_MAX_KEYS]; Constraint *fkconstraint; + bool attach_it; ClonedConstraint *newc; Oid constrOid; ObjectAddress parentAddr, childAddr; int nelem; + ListCell *cell; int i; ArrayType *arr; Datum datum; @@ -539,6 +544,66 @@ CloneForeignKeyConstraints(Oid parentId, Oid relationId, List **cloned) elog(ERROR, "conffeqop is not a 1-D OID array"); memcpy(conffeqop, ARR_DATA_PTR(arr), nelem * sizeof(Oid)); + /* + * Before creating a new constraint, see whether any existing FKs are + * fit for the purpose. If it is, attach the parent constraint to it + * instead of creating a new constraint. This way we avoid the + * expensive verification step and don't end up with a duplicate FK. + */ + attach_it = false; + foreach (cell, partFKs) + { + ForeignKeyCacheInfo *fk = lfirst_node(ForeignKeyCacheInfo, cell); + + if (fk->confrelid == constrForm->confrelid && + fk->nkeys == nelem) + { + Form_pg_constraint partConstr; + HeapTuple partcontup; + + /* Do some quick & easy initial checks */ + attach_it = true; + for (i = 0; i < nelem; i++) + { + if (fk->conkey[i] != mapped_conkey[i] || + fk->confkey[i] != confkey[i] || + fk->conpfeqop[i] != conpfeqop[i]) + { + attach_it = false; + break; + } + } + + /* looks good ... do some more extensive checks */ + partcontup = SearchSysCache1(CONSTROID, + ObjectIdGetDatum(fk->conoid)); + if (!partcontup) + elog(ERROR, "cache lookup failed for constraint %u", + fk->conoid); + partConstr = (Form_pg_constraint) GETSTRUCT(partcontup); + if (partConstr->condeferrable != constrForm->condeferrable || + partConstr->condeferred != constrForm->condeferred || + partConstr->confupdtype != constrForm->confupdtype || + partConstr->confdeltype != constrForm->confdeltype || + partConstr->confmatchtype != constrForm->confmatchtype) + { + attach_it = false; + ReleaseSysCache(partcontup); + break; + } + + ReleaseSysCache(partcontup); + if (attach_it) + { + ConstraintSetParentConstraint(fk->conoid, + HeapTupleGetOid(tuple)); + break; + } + } + } + if (attach_it) + continue; + constrOid = CreateConstraintEntry(NameStr(constrForm->conname), constrForm->connamespace, @@ -602,6 +667,7 @@ CloneForeignKeyConstraints(Oid parentId, Oid relationId, List **cloned) systable_endscan(scan); pfree(attmap); + list_free_deep(partFKs); if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) { @@ -1028,17 +1094,33 @@ ConstraintSetParentConstraint(Oid childConstrId, Oid parentConstrId) elog(ERROR, "cache lookup failed for constraint %u", childConstrId); newtup = heap_copytuple(tuple); constrForm = (Form_pg_constraint) GETSTRUCT(newtup); - constrForm->conislocal = false; - constrForm->coninhcount++; - constrForm->conparentid = parentConstrId; - CatalogTupleUpdate(constrRel, &tuple->t_self, newtup); + if (OidIsValid(parentConstrId)) + { + constrForm->conislocal = false; + constrForm->coninhcount++; + constrForm->conparentid = parentConstrId; + + CatalogTupleUpdate(constrRel, &tuple->t_self, newtup); + + ObjectAddressSet(referenced, ConstraintRelationId, parentConstrId); + ObjectAddressSet(depender, ConstraintRelationId, childConstrId); + + recordDependencyOn(&depender, &referenced, DEPENDENCY_INTERNAL_AUTO); + } + else + { + constrForm->coninhcount--; + if (constrForm->coninhcount <= 0) + constrForm->conislocal = true; + constrForm->conparentid = InvalidOid; + + deleteDependencyRecordsForClass(ConstraintRelationId, childConstrId, + ConstraintRelationId, + DEPENDENCY_INTERNAL_AUTO); + CatalogTupleUpdate(constrRel, &tuple->t_self, newtup); + } + ReleaseSysCache(tuple); - - ObjectAddressSet(referenced, ConstraintRelationId, parentConstrId); - ObjectAddressSet(depender, ConstraintRelationId, childConstrId); - - recordDependencyOn(&depender, &referenced, DEPENDENCY_INTERNAL_AUTO); - heap_close(constrRel, RowExclusiveLock); } diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index f988c16659..2045feb17e 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -14607,6 +14607,7 @@ ATExecDetachPartition(Relation rel, RangeVar *name) ObjectAddress address; Oid defaultPartOid; List *indexes; + List *fks; ListCell *cell; /* @@ -14682,6 +14683,24 @@ ATExecDetachPartition(Relation rel, RangeVar *name) } heap_close(classRel, RowExclusiveLock); + /* Detach foreign keys */ + + fks = copyObject(RelationGetFKeyList(partRel)); + foreach (cell, fks) + { + ForeignKeyCacheInfo *fk = lfirst(cell); + HeapTuple contup; + + contup = SearchSysCache1(CONSTROID, ObjectIdGetDatum(fk->conoid)); + if (!contup) + elog(ERROR, "cache lookup failed for constraint %u", fk->conoid); + + ConstraintSetParentConstraint(fk->conoid, InvalidOid); + + ReleaseSysCache(contup); + } + list_free_deep(fks); + /* * Invalidate the parent's relcache so that the partition is no longer * included in its partition descriptor. diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 39618323fc..648758de4a 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -4746,6 +4746,7 @@ _copyForeignKeyCacheInfo(const ForeignKeyCacheInfo *from) { ForeignKeyCacheInfo *newnode = makeNode(ForeignKeyCacheInfo); + COPY_SCALAR_FIELD(conoid); COPY_SCALAR_FIELD(conrelid); COPY_SCALAR_FIELD(confrelid); COPY_SCALAR_FIELD(nkeys); diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c index 9a169f7c6c..6c3dad9ab3 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -3634,6 +3634,7 @@ _outForeignKeyCacheInfo(StringInfo str, const ForeignKeyCacheInfo *node) WRITE_NODE_TYPE("FOREIGNKEYCACHEINFO"); + WRITE_OID_FIELD(conoid); WRITE_OID_FIELD(conrelid); WRITE_OID_FIELD(confrelid); WRITE_INT_FIELD(nkeys); diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index a4fc001103..1f29956767 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -4144,6 +4144,7 @@ RelationGetFKeyList(Relation relation) continue; info = makeNode(ForeignKeyCacheInfo); + info->conoid = HeapTupleGetOid(htup); info->conrelid = constraint->conrelid; info->confrelid = constraint->confrelid; diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index 6ecbdb6294..84469f5715 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -202,12 +202,13 @@ typedef struct RelationData * The per-FK-column arrays can be fixed-size because we allow at most * INDEX_MAX_KEYS columns in a foreign key constraint. * - * Currently, we only cache fields of interest to the planner, but the - * set of fields could be expanded in future. + * Currently, we mostly cache fields of interest to the planner, but the set + * of fields has already grown the constraint OID for other uses. */ typedef struct ForeignKeyCacheInfo { NodeTag type; + Oid conoid; /* oid of the constraint itself */ Oid conrelid; /* relation constrained by the foreign key */ Oid confrelid; /* relation referenced by the foreign key */ int nkeys; /* number of columns in the foreign key */ diff --git a/src/test/regress/expected/foreign_key.out b/src/test/regress/expected/foreign_key.out index 4e5cb8901e..e252423e4c 100644 --- a/src/test/regress/expected/foreign_key.out +++ b/src/test/regress/expected/foreign_key.out @@ -1648,6 +1648,41 @@ SELECT * FROM fk_partitioned_fk WHERE a = 142857; -- verify that DROP works DROP TABLE fk_partitioned_fk_2; +-- Test behavior of the constraint together with attaching and detaching +-- partitions. +CREATE TABLE fk_partitioned_fk_2 PARTITION OF fk_partitioned_fk FOR VALUES IN (1500,1502); +ALTER TABLE fk_partitioned_fk DETACH PARTITION fk_partitioned_fk_2; +BEGIN; +DROP TABLE fk_partitioned_fk; +-- constraint should still be there +\d fk_partitioned_fk_2; + Table "public.fk_partitioned_fk_2" + Column | Type | Collation | Nullable | Default +--------+---------+-----------+----------+--------- + a | integer | | | 2501 + b | integer | | | 142857 +Foreign-key constraints: + "fk_partitioned_fk_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE + +ROLLBACK; +ALTER TABLE fk_partitioned_fk ATTACH PARTITION fk_partitioned_fk_2 FOR VALUES IN (1500,1502); +DROP TABLE fk_partitioned_fk_2; +CREATE TABLE fk_partitioned_fk_2 (c text, a int, b int, + FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk ON UPDATE CASCADE ON DELETE CASCADE); +ALTER TABLE fk_partitioned_fk_2 DROP COLUMN c; +ALTER TABLE fk_partitioned_fk ATTACH PARTITION fk_partitioned_fk_2 FOR VALUES IN (1500,1502); +-- should have only one constraint +\d fk_partitioned_fk_2 + Table "public.fk_partitioned_fk_2" + Column | Type | Collation | Nullable | Default +--------+---------+-----------+----------+--------- + a | integer | | | + b | integer | | | +Partition of: fk_partitioned_fk FOR VALUES IN (1500, 1502) +Foreign-key constraints: + "fk_partitioned_fk_2_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE + +DROP TABLE fk_partitioned_fk_2; -- verify that attaching a table checks that the existing data satisfies the -- constraint CREATE TABLE fk_partitioned_fk_2 (a int, b int) PARTITION BY RANGE (b); diff --git a/src/test/regress/sql/foreign_key.sql b/src/test/regress/sql/foreign_key.sql index 6fcb5dfb4e..b01d5cef12 100644 --- a/src/test/regress/sql/foreign_key.sql +++ b/src/test/regress/sql/foreign_key.sql @@ -1226,6 +1226,25 @@ SELECT * FROM fk_partitioned_fk WHERE a = 142857; -- verify that DROP works DROP TABLE fk_partitioned_fk_2; +-- Test behavior of the constraint together with attaching and detaching +-- partitions. +CREATE TABLE fk_partitioned_fk_2 PARTITION OF fk_partitioned_fk FOR VALUES IN (1500,1502); +ALTER TABLE fk_partitioned_fk DETACH PARTITION fk_partitioned_fk_2; +BEGIN; +DROP TABLE fk_partitioned_fk; +-- constraint should still be there +\d fk_partitioned_fk_2; +ROLLBACK; +ALTER TABLE fk_partitioned_fk ATTACH PARTITION fk_partitioned_fk_2 FOR VALUES IN (1500,1502); +DROP TABLE fk_partitioned_fk_2; +CREATE TABLE fk_partitioned_fk_2 (b int, c text, a int, + FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk ON UPDATE CASCADE ON DELETE CASCADE); +ALTER TABLE fk_partitioned_fk_2 DROP COLUMN c; +ALTER TABLE fk_partitioned_fk ATTACH PARTITION fk_partitioned_fk_2 FOR VALUES IN (1500,1502); +-- should have only one constraint +\d fk_partitioned_fk_2 +DROP TABLE fk_partitioned_fk_2; + -- verify that attaching a table checks that the existing data satisfies the -- constraint CREATE TABLE fk_partitioned_fk_2 (a int, b int) PARTITION BY RANGE (b);