From: Alvaro Herrera <alvherre@2ndquadrant.com>
To: Pg Hackers <pgsql-hackers@postgresql.org>
Subject: Re: partitioned tables referenced by FKs
Date: Fri, 30 Nov 2018 16:12:16 -0300
Message-ID: <20181130191216.5xcxcsx3ascgqayv@alvherre.pgsql> (raw)
In-Reply-To: <20181102234158.735b3fevta63msbj@alvherre.pgsql>
Here's a more credible version of this patch series.
0001 refactors some duplicative code that interprets a pg_constraint row
for a foreign key back into a Constraint node. Only what is necessary
for current features is read.
0002 moves some code that I added in 3de241dba86f to
src/backend/catalog/pg_constraint.c so that it appears in tablecmds.c
instead. After developing the current patch I realized that the latter
is its natural home.
0003 changes the shape of a loop in deleteObjectsInList, so that I can
add object type-specific functions to be called before deleting an
object. This is needed by 0004 and makes no sense on its own; I would
probably push both together, but splitting it like this makes it very
obvious what it is I'm doing.
0004 adds the feature itself
In 0004 there's also a new function "index_get_partition" which allows
to simplify some code I added in 8b08f7d4820f. I will probably split it
up and commit separately because it's an obvious code beautify.
0005 modifies psql to show the constraint in a different way, which
makes more sense overall (rather than show the "internal" constraint
that concerns the partition, show the top-level contraint that concerns
the ancestor table on which it is defined. This includes naming the
table in which the constraint is defined).
There's one half of 0005 that I think should be applied to pg11, because
when partitions have foreign keys, the display looks a bit odd
currently. The other half I would probably merge into 0004 instead of
committing separately. Even if not backpatched, it makes sense to apply
ahead of the rest because it changes expected output for a regression
test.
--
Álvaro Herrera https://www.2ndQuadrant.com/
PostgreSQL Development, 24x7 Support, Remote DBA, Training & Services
Attachments:
[text/x-diff] 0001-Refactor-DeconstructConstraintRow.patch (15.9K, ../20181130191216.5xcxcsx3ascgqayv@alvherre.pgsql/2-0001-Refactor-DeconstructConstraintRow.patch)
download | inline diff:
From 563447095c3884a00ecfd933027524e252cf35f7 Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Mon, 12 Nov 2018 16:05:16 -0300
Subject: [PATCH 1/5] Refactor DeconstructConstraintRow
---
src/backend/catalog/pg_constraint.c | 200 ++++++++++++++++++++----------------
src/backend/utils/adt/ri_triggers.c | 89 ++--------------
src/backend/utils/cache/relcache.c | 61 +----------
src/include/catalog/pg_constraint.h | 3 +
4 files changed, 125 insertions(+), 228 deletions(-)
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.cindex a8194b02fa..f71d89ff17 100644--- a/src/backend/catalog/pg_constraint.c+++ b/src/backend/catalog/pg_constraint.c@@ -446,14 +446,11 @@ static void
clone_fk_constraints(Relation pg_constraint, Relation parentRel,
Relation partRel, List *clone, List **cloned)
{
- TupleDesc tupdesc;
AttrNumber *attmap;
List *partFKs;
List *subclone = NIL;
ListCell *cell;
- tupdesc = RelationGetDescr(pg_constraint);-
/*
* The constraint key may differ, if the columns in the partition are
* different. This map is used to convert them.
@@ -483,9 +480,6 @@ clone_fk_constraints(Relation pg_constraint, Relation parentRel,
int nelem;
ListCell *cell;
int i;
- ArrayType *arr;- Datum datum;- bool isnull;
tuple = SearchSysCache1(CONSTROID, parentConstrOid);
if (!tuple)
@@ -502,93 +496,11 @@ clone_fk_constraints(Relation pg_constraint, Relation parentRel,
ObjectAddressSet(parentAddr, ConstraintRelationId, parentConstrOid);
- datum = fastgetattr(tuple, Anum_pg_constraint_conkey,- tupdesc, &isnull);- if (isnull)- elog(ERROR, "null conkey");- arr = DatumGetArrayTypeP(datum);- nelem = ARR_DIMS(arr)[0];- if (ARR_NDIM(arr) != 1 ||- nelem < 1 ||- nelem > INDEX_MAX_KEYS ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != INT2OID)- elog(ERROR, "conkey is not a 1-D smallint array");- memcpy(conkey, ARR_DATA_PTR(arr), nelem * sizeof(AttrNumber));-+ DeconstructConstraintRow(tuple, &nelem, conkey, confkey,+ conpfeqop, conppeqop, conffeqop);
for (i = 0; i < nelem; i++)
mapped_conkey[i] = attmap[conkey[i] - 1];
- datum = fastgetattr(tuple, Anum_pg_constraint_confkey,- tupdesc, &isnull);- if (isnull)- elog(ERROR, "null confkey");- arr = DatumGetArrayTypeP(datum);- nelem = ARR_DIMS(arr)[0];- if (ARR_NDIM(arr) != 1 ||- nelem < 1 ||- nelem > INDEX_MAX_KEYS ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != INT2OID)- elog(ERROR, "confkey is not a 1-D smallint array");- memcpy(confkey, ARR_DATA_PTR(arr), nelem * sizeof(AttrNumber));-- datum = fastgetattr(tuple, Anum_pg_constraint_conpfeqop,- tupdesc, &isnull);- if (isnull)- elog(ERROR, "null conpfeqop");- arr = DatumGetArrayTypeP(datum);- nelem = ARR_DIMS(arr)[0];- if (ARR_NDIM(arr) != 1 ||- nelem < 1 ||- nelem > INDEX_MAX_KEYS ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != OIDOID)- elog(ERROR, "conpfeqop is not a 1-D OID array");- memcpy(conpfeqop, ARR_DATA_PTR(arr), nelem * sizeof(Oid));-- datum = fastgetattr(tuple, Anum_pg_constraint_conpfeqop,- tupdesc, &isnull);- if (isnull)- elog(ERROR, "null conpfeqop");- arr = DatumGetArrayTypeP(datum);- nelem = ARR_DIMS(arr)[0];- if (ARR_NDIM(arr) != 1 ||- nelem < 1 ||- nelem > INDEX_MAX_KEYS ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != OIDOID)- elog(ERROR, "conpfeqop is not a 1-D OID array");- memcpy(conpfeqop, ARR_DATA_PTR(arr), nelem * sizeof(Oid));-- datum = fastgetattr(tuple, Anum_pg_constraint_conppeqop,- tupdesc, &isnull);- if (isnull)- elog(ERROR, "null conppeqop");- arr = DatumGetArrayTypeP(datum);- nelem = ARR_DIMS(arr)[0];- if (ARR_NDIM(arr) != 1 ||- nelem < 1 ||- nelem > INDEX_MAX_KEYS ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != OIDOID)- elog(ERROR, "conppeqop is not a 1-D OID array");- memcpy(conppeqop, ARR_DATA_PTR(arr), nelem * sizeof(Oid));-- datum = fastgetattr(tuple, Anum_pg_constraint_conffeqop,- tupdesc, &isnull);- if (isnull)- elog(ERROR, "null conffeqop");- arr = DatumGetArrayTypeP(datum);- nelem = ARR_DIMS(arr)[0];- if (ARR_NDIM(arr) != 1 ||- nelem < 1 ||- nelem > INDEX_MAX_KEYS ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != OIDOID)- 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 one is, attach the parent constraint to it,
@@ -1531,6 +1443,114 @@ get_primary_key_attnos(Oid relid, bool deferrableOk, Oid *constraintOid)
}
/*
+ * Extract data from the pg_constraint row of a foreign-key constraint.+ *+ * The last three can be passed as NULL.+ */+void+DeconstructConstraintRow(HeapTuple tuple, int *numfks,+ AttrNumber *conkey, AttrNumber *confkey,+ Oid *pf_eq_oprs, Oid *pp_eq_oprs, Oid *ff_eq_oprs)+{+ Oid constrId;+ Datum adatum;+ bool isNull;+ ArrayType *arr;+ int numkeys;++ constrId = ((Form_pg_constraint) GETSTRUCT(tuple))->oid;++ /*+ * We expect the arrays to be 1-D arrays of the right types; verify that.+ * We don't need to use deconstruct_array() since the array data is just+ * going to look like a C array of values.+ */+ adatum = SysCacheGetAttr(CONSTROID, tuple,+ Anum_pg_constraint_conkey, &isNull);+ if (isNull)+ elog(ERROR, "null conkey for constraint %u", constrId);+ arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */+ if (ARR_NDIM(arr) != 1 ||+ ARR_HASNULL(arr) ||+ ARR_ELEMTYPE(arr) != INT2OID)+ elog(ERROR, "conkey is not a 1-D smallint array");+ numkeys = ARR_DIMS(arr)[0];+ if (numkeys <= 0 || numkeys > INDEX_MAX_KEYS)+ elog(ERROR, "foreign key constraint cannot have %d columns", numkeys);+ memcpy(conkey, ARR_DATA_PTR(arr), numkeys * sizeof(int16));+ if ((Pointer) arr != DatumGetPointer(adatum))+ pfree(arr); /* free de-toasted copy, if any */++ adatum = SysCacheGetAttr(CONSTROID, tuple,+ Anum_pg_constraint_confkey, &isNull);+ if (isNull)+ elog(ERROR, "null confkey for constraint %u", constrId);+ arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */+ if (ARR_NDIM(arr) != 1 ||+ ARR_DIMS(arr)[0] != numkeys ||+ ARR_HASNULL(arr) ||+ ARR_ELEMTYPE(arr) != INT2OID)+ elog(ERROR, "confkey is not a 1-D smallint array");+ memcpy(confkey, ARR_DATA_PTR(arr), numkeys * sizeof(int16));+ if ((Pointer) arr != DatumGetPointer(adatum))+ pfree(arr); /* free de-toasted copy, if any */++ if (pf_eq_oprs)+ {+ adatum = SysCacheGetAttr(CONSTROID, tuple,+ Anum_pg_constraint_conpfeqop, &isNull);+ if (isNull)+ elog(ERROR, "null conpfeqop for constraint %u", constrId);+ arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */+ /* see TryReuseForeignKey if you change the test below */+ if (ARR_NDIM(arr) != 1 ||+ ARR_DIMS(arr)[0] != numkeys ||+ ARR_HASNULL(arr) ||+ ARR_ELEMTYPE(arr) != OIDOID)+ elog(ERROR, "conpfeqop is not a 1-D Oid array");+ memcpy(pf_eq_oprs, ARR_DATA_PTR(arr), numkeys * sizeof(Oid));+ if ((Pointer) arr != DatumGetPointer(adatum))+ pfree(arr); /* free de-toasted copy, if any */+ }++ if (pp_eq_oprs)+ {+ adatum = SysCacheGetAttr(CONSTROID, tuple,+ Anum_pg_constraint_conppeqop, &isNull);+ if (isNull)+ elog(ERROR, "null conppeqop for constraint %u", constrId);+ arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */+ if (ARR_NDIM(arr) != 1 ||+ ARR_DIMS(arr)[0] != numkeys ||+ ARR_HASNULL(arr) ||+ ARR_ELEMTYPE(arr) != OIDOID)+ elog(ERROR, "conppeqop is not a 1-D Oid array");+ memcpy(pp_eq_oprs, ARR_DATA_PTR(arr), numkeys * sizeof(Oid));+ if ((Pointer) arr != DatumGetPointer(adatum))+ pfree(arr); /* free de-toasted copy, if any */+ }++ if (ff_eq_oprs)+ {+ adatum = SysCacheGetAttr(CONSTROID, tuple,+ Anum_pg_constraint_conffeqop, &isNull);+ if (isNull)+ elog(ERROR, "null conffeqop for constraint %u", constrId);+ arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */+ if (ARR_NDIM(arr) != 1 ||+ ARR_DIMS(arr)[0] != numkeys ||+ ARR_HASNULL(arr) ||+ ARR_ELEMTYPE(arr) != OIDOID)+ elog(ERROR, "conffeqop is not a 1-D Oid array");+ memcpy(ff_eq_oprs, ARR_DATA_PTR(arr), numkeys * sizeof(Oid));+ if ((Pointer) arr != DatumGetPointer(adatum))+ pfree(arr); /* free de-toasted copy, if any */+ }++ *numfks = numkeys;+}++/*
* Determine whether a relation can be proven functionally dependent on
* a set of grouping columns. If so, return true and add the pg_constraint
* OIDs of the constraints needed for the proof to the *constraintDeps list.
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.cindex cdda860e73..b9aef07c1a 100644--- a/src/backend/utils/adt/ri_triggers.c+++ b/src/backend/utils/adt/ri_triggers.c@@ -2188,10 +2188,6 @@ ri_LoadConstraintInfo(Oid constraintOid)
bool found;
HeapTuple tup;
Form_pg_constraint conForm;
- Datum adatum;- bool isNull;- ArrayType *arr;- int numkeys;
/*
* On the first call initialize the hashtable
@@ -2233,84 +2229,13 @@ ri_LoadConstraintInfo(Oid constraintOid)
riinfo->confdeltype = conForm->confdeltype;
riinfo->confmatchtype = conForm->confmatchtype;
- /*- * We expect the arrays to be 1-D arrays of the right types; verify that.- * We don't need to use deconstruct_array() since the array data is just- * going to look like a C array of values.- */- adatum = SysCacheGetAttr(CONSTROID, tup,- Anum_pg_constraint_conkey, &isNull);- if (isNull)- elog(ERROR, "null conkey for constraint %u", constraintOid);- arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */- if (ARR_NDIM(arr) != 1 ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != INT2OID)- elog(ERROR, "conkey is not a 1-D smallint array");- numkeys = ARR_DIMS(arr)[0];- if (numkeys <= 0 || numkeys > RI_MAX_NUMKEYS)- elog(ERROR, "foreign key constraint cannot have %d columns", numkeys);- riinfo->nkeys = numkeys;- memcpy(riinfo->fk_attnums, ARR_DATA_PTR(arr), numkeys * sizeof(int16));- if ((Pointer) arr != DatumGetPointer(adatum))- pfree(arr); /* free de-toasted copy, if any */-- adatum = SysCacheGetAttr(CONSTROID, tup,- Anum_pg_constraint_confkey, &isNull);- if (isNull)- elog(ERROR, "null confkey for constraint %u", constraintOid);- arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */- if (ARR_NDIM(arr) != 1 ||- ARR_DIMS(arr)[0] != numkeys ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != INT2OID)- elog(ERROR, "confkey is not a 1-D smallint array");- memcpy(riinfo->pk_attnums, ARR_DATA_PTR(arr), numkeys * sizeof(int16));- if ((Pointer) arr != DatumGetPointer(adatum))- pfree(arr); /* free de-toasted copy, if any */-- adatum = SysCacheGetAttr(CONSTROID, tup,- Anum_pg_constraint_conpfeqop, &isNull);- if (isNull)- elog(ERROR, "null conpfeqop for constraint %u", constraintOid);- arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */- /* see TryReuseForeignKey if you change the test below */- if (ARR_NDIM(arr) != 1 ||- ARR_DIMS(arr)[0] != numkeys ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != OIDOID)- elog(ERROR, "conpfeqop is not a 1-D Oid array");- memcpy(riinfo->pf_eq_oprs, ARR_DATA_PTR(arr), numkeys * sizeof(Oid));- if ((Pointer) arr != DatumGetPointer(adatum))- pfree(arr); /* free de-toasted copy, if any */-- adatum = SysCacheGetAttr(CONSTROID, tup,- Anum_pg_constraint_conppeqop, &isNull);- if (isNull)- elog(ERROR, "null conppeqop for constraint %u", constraintOid);- arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */- if (ARR_NDIM(arr) != 1 ||- ARR_DIMS(arr)[0] != numkeys ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != OIDOID)- elog(ERROR, "conppeqop is not a 1-D Oid array");- memcpy(riinfo->pp_eq_oprs, ARR_DATA_PTR(arr), numkeys * sizeof(Oid));- if ((Pointer) arr != DatumGetPointer(adatum))- pfree(arr); /* free de-toasted copy, if any */-- adatum = SysCacheGetAttr(CONSTROID, tup,- Anum_pg_constraint_conffeqop, &isNull);- if (isNull)- elog(ERROR, "null conffeqop for constraint %u", constraintOid);- arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */- if (ARR_NDIM(arr) != 1 ||- ARR_DIMS(arr)[0] != numkeys ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != OIDOID)- elog(ERROR, "conffeqop is not a 1-D Oid array");- memcpy(riinfo->ff_eq_oprs, ARR_DATA_PTR(arr), numkeys * sizeof(Oid));- if ((Pointer) arr != DatumGetPointer(adatum))- pfree(arr); /* free de-toasted copy, if any */+ DeconstructConstraintRow(tup,+ &riinfo->nkeys,+ riinfo->fk_attnums,+ riinfo->pk_attnums,+ riinfo->pf_eq_oprs,+ riinfo->pp_eq_oprs,+ riinfo->ff_eq_oprs);
ReleaseSysCache(tup);
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.cindex c3071db1cd..1dda178c5a 100644--- a/src/backend/utils/cache/relcache.c+++ b/src/backend/utils/cache/relcache.c@@ -4124,10 +4124,6 @@ RelationGetFKeyList(Relation relation)
{
Form_pg_constraint constraint = (Form_pg_constraint) GETSTRUCT(htup);
ForeignKeyCacheInfo *info;
- Datum adatum;- bool isnull;- ArrayType *arr;- int nelem;
/* consider only foreign keys */
if (constraint->contype != CONSTRAINT_FOREIGN)
@@ -4138,58 +4134,11 @@ RelationGetFKeyList(Relation relation)
info->conrelid = constraint->conrelid;
info->confrelid = constraint->confrelid;
- /* Extract data from conkey field */- adatum = fastgetattr(htup, Anum_pg_constraint_conkey,- conrel->rd_att, &isnull);- if (isnull)- elog(ERROR, "null conkey for rel %s",- RelationGetRelationName(relation));-- arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */- nelem = ARR_DIMS(arr)[0];- if (ARR_NDIM(arr) != 1 ||- nelem < 1 ||- nelem > INDEX_MAX_KEYS ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != INT2OID)- elog(ERROR, "conkey is not a 1-D smallint array");-- info->nkeys = nelem;- memcpy(info->conkey, ARR_DATA_PTR(arr), nelem * sizeof(AttrNumber));-- /* Likewise for confkey */- adatum = fastgetattr(htup, Anum_pg_constraint_confkey,- conrel->rd_att, &isnull);- if (isnull)- elog(ERROR, "null confkey for rel %s",- RelationGetRelationName(relation));-- arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */- nelem = ARR_DIMS(arr)[0];- if (ARR_NDIM(arr) != 1 ||- nelem != info->nkeys ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != INT2OID)- elog(ERROR, "confkey is not a 1-D smallint array");-- memcpy(info->confkey, ARR_DATA_PTR(arr), nelem * sizeof(AttrNumber));-- /* Likewise for conpfeqop */- adatum = fastgetattr(htup, Anum_pg_constraint_conpfeqop,- conrel->rd_att, &isnull);- if (isnull)- elog(ERROR, "null conpfeqop for rel %s",- RelationGetRelationName(relation));-- arr = DatumGetArrayTypeP(adatum); /* ensure not toasted */- nelem = ARR_DIMS(arr)[0];- if (ARR_NDIM(arr) != 1 ||- nelem != info->nkeys ||- ARR_HASNULL(arr) ||- ARR_ELEMTYPE(arr) != OIDOID)- elog(ERROR, "conpfeqop is not a 1-D OID array");-- memcpy(info->conpfeqop, ARR_DATA_PTR(arr), nelem * sizeof(Oid));+ DeconstructConstraintRow(htup, &info->nkeys,+ info->conkey,+ info->confkey,+ info->conpfeqop,+ NULL, NULL);
/* Add FK's node to the result list */
result = lappend(result, info);
diff --git a/src/include/catalog/pg_constraint.h b/src/include/catalog/pg_constraint.hindex 0e4007389b..8405fe2eea 100644--- a/src/include/catalog/pg_constraint.h+++ b/src/include/catalog/pg_constraint.h@@ -251,6 +251,9 @@ extern Oid get_relation_idx_constraint_oid(Oid relationId, Oid indexId);
extern Bitmapset *get_primary_key_attnos(Oid relid, bool deferrableOk,
Oid *constraintOid);
+extern void DeconstructConstraintRow(HeapTuple tuple, int *numfks,+ AttrNumber *conkey, AttrNumber *confkey,+ Oid *pf_eq_oprs, Oid *pp_eq_oprs, Oid *ff_eq_oprs);
extern bool check_functional_grouping(Oid relid,
Index varno, Index varlevelsup,
--
2.11.0
[text/x-diff] 0002-move-CloneForeignKeyConstraints-to-tablecmds.c.patch (20.6K, ../20181130191216.5xcxcsx3ascgqayv@alvherre.pgsql/3-0002-move-CloneForeignKeyConstraints-to-tablecmds.c.patch)
download | inline diff:
From eb68d572c815c34500f6ed720860f24386afe9cd Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Mon, 12 Nov 2018 16:18:45 -0300
Subject: [PATCH 2/5] move CloneForeignKeyConstraints to tablecmds.c
---
src/backend/catalog/pg_constraint.c | 304 -----------------------------------
src/backend/commands/tablecmds.c | 306 ++++++++++++++++++++++++++++++++++++
src/include/catalog/pg_constraint.h | 3 -
3 files changed, 306 insertions(+), 307 deletions(-)
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.cindex f71d89ff17..2833e45cff 100644--- a/src/backend/catalog/pg_constraint.c+++ b/src/backend/catalog/pg_constraint.c@@ -39,10 +39,6 @@
#include "utils/tqual.h"
-static void clone_fk_constraints(Relation pg_constraint, Relation parentRel,- Relation partRel, List *clone, List **cloned);--
/*
* CreateConstraintEntry
* Create a constraint table entry.
@@ -378,306 +374,6 @@ CreateConstraintEntry(const char *constraintName,
}
/*
- * CloneForeignKeyConstraints- * Clone foreign keys from a partitioned table to a newly acquired- * partition.- *- * relationId is a partition of parentId, so we can be certain that it has the- * same columns with the same datatypes. The columns may be in different- * order, though.- *- * The *cloned list is appended ClonedConstraint elements describing what was- * created.- */-void-CloneForeignKeyConstraints(Oid parentId, Oid relationId, List **cloned)-{- Relation pg_constraint;- Relation parentRel;- Relation rel;- ScanKeyData key;- SysScanDesc scan;- HeapTuple tuple;- List *clone = NIL;-- parentRel = heap_open(parentId, NoLock); /* already got lock */- /* see ATAddForeignKeyConstraint about lock level */- rel = heap_open(relationId, AccessExclusiveLock);- pg_constraint = heap_open(ConstraintRelationId, RowShareLock);-- /* Obtain the list of constraints to clone or attach */- ScanKeyInit(&key,- Anum_pg_constraint_conrelid, BTEqualStrategyNumber,- F_OIDEQ, ObjectIdGetDatum(parentId));- scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId, true,- NULL, 1, &key);- while ((tuple = systable_getnext(scan)) != NULL)- {- Oid oid = ((Form_pg_constraint) GETSTRUCT(tuple))->oid;-- clone = lappend_oid(clone, oid);- }- systable_endscan(scan);-- /* Do the actual work, recursing to partitions as needed */- clone_fk_constraints(pg_constraint, parentRel, rel, clone, cloned);-- /* We're done. Clean up */- heap_close(parentRel, NoLock);- heap_close(rel, NoLock); /* keep lock till commit */- heap_close(pg_constraint, RowShareLock);-}--/*- * clone_fk_constraints- * Recursive subroutine for CloneForeignKeyConstraints- *- * Clone the given list of FK constraints when a partition is attached.- *- * When cloning foreign keys to a partition, it may happen that equivalent- * constraints already exist in the partition for some of them. We can skip- * creating a clone in that case, and instead just attach the existing- * constraint to the one in the parent.- *- * This function recurses to partitions, if the new partition is partitioned;- * of course, only do this for FKs that were actually cloned.- */-static void-clone_fk_constraints(Relation pg_constraint, Relation parentRel,- Relation partRel, List *clone, List **cloned)-{- AttrNumber *attmap;- List *partFKs;- List *subclone = NIL;- ListCell *cell;-- /*- * The constraint key may differ, if the columns in the partition are- * different. This map is used to convert them.- */- attmap = convert_tuples_by_name_map(RelationGetDescr(partRel),- RelationGetDescr(parentRel),- gettext_noop("could not convert row type"));-- partFKs = copyObject(RelationGetFKeyList(partRel));-- foreach(cell, clone)- {- Oid parentConstrOid = lfirst_oid(cell);- Form_pg_constraint constrForm;- HeapTuple tuple;- AttrNumber conkey[INDEX_MAX_KEYS];- AttrNumber mapped_conkey[INDEX_MAX_KEYS];- AttrNumber confkey[INDEX_MAX_KEYS];- Oid conpfeqop[INDEX_MAX_KEYS];- Oid conppeqop[INDEX_MAX_KEYS];- Oid conffeqop[INDEX_MAX_KEYS];- Constraint *fkconstraint;- bool attach_it;- Oid constrOid;- ObjectAddress parentAddr,- childAddr;- int nelem;- ListCell *cell;- int i;-- tuple = SearchSysCache1(CONSTROID, parentConstrOid);- if (!tuple)- elog(ERROR, "cache lookup failed for constraint %u",- parentConstrOid);- constrForm = (Form_pg_constraint) GETSTRUCT(tuple);-- /* only foreign keys */- if (constrForm->contype != CONSTRAINT_FOREIGN)- {- ReleaseSysCache(tuple);- continue;- }-- ObjectAddressSet(parentAddr, ConstraintRelationId, parentConstrOid);-- DeconstructConstraintRow(tuple, &nelem, conkey, confkey,- conpfeqop, conppeqop, conffeqop);- for (i = 0; i < nelem; i++)- mapped_conkey[i] = attmap[conkey[i] - 1];-- /*- * Before creating a new constraint, see whether any existing FKs are- * fit for the purpose. If one is, attach the parent constraint to it,- * and don't clone anything. This way we avoid the expensive- * verification step and don't end up with a duplicate FK. This also- * means we don't consider this constraint when recursing to- * partitions.- */- attach_it = false;- foreach(cell, partFKs)- {- ForeignKeyCacheInfo *fk = lfirst_node(ForeignKeyCacheInfo, cell);- Form_pg_constraint partConstr;- HeapTuple partcontup;-- attach_it = true;-- /*- * Do some quick & easy initial checks. If any of these fail, we- * cannot use this constraint, but keep looking.- */- if (fk->confrelid != constrForm->confrelid || fk->nkeys != nelem)- {- attach_it = false;- continue;- }- 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;- }- }- if (!attach_it)- continue;-- /*- * Looks good so far; do some more extensive checks. Presumably- * the check for 'convalidated' could be dropped, since we don't- * really care about that, but let's be careful for now.- */- 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 (OidIsValid(partConstr->conparentid) ||- !partConstr->convalidated ||- partConstr->condeferrable != constrForm->condeferrable ||- partConstr->condeferred != constrForm->condeferred ||- partConstr->confupdtype != constrForm->confupdtype ||- partConstr->confdeltype != constrForm->confdeltype ||- partConstr->confmatchtype != constrForm->confmatchtype)- {- ReleaseSysCache(partcontup);- attach_it = false;- continue;- }-- ReleaseSysCache(partcontup);-- /* looks good! Attach this constraint */- ConstraintSetParentConstraint(fk->conoid, constrForm->oid);- CommandCounterIncrement();- attach_it = true;- break;- }-- /*- * If we attached to an existing constraint, there is no need to- * create a new one. In fact, there's no need to recurse for this- * constraint to partitions, either.- */- if (attach_it)- {- ReleaseSysCache(tuple);- continue;- }-- constrOid =- CreateConstraintEntry(NameStr(constrForm->conname),- constrForm->connamespace,- CONSTRAINT_FOREIGN,- constrForm->condeferrable,- constrForm->condeferred,- constrForm->convalidated,- constrForm->oid,- RelationGetRelid(partRel),- mapped_conkey,- nelem,- nelem,- InvalidOid, /* not a domain constraint */- constrForm->conindid, /* same index */- constrForm->confrelid, /* same foreign rel */- confkey,- conpfeqop,- conppeqop,- conffeqop,- nelem,- constrForm->confupdtype,- constrForm->confdeltype,- constrForm->confmatchtype,- NULL,- NULL,- NULL,- false,- 1, false, true);- subclone = lappend_oid(subclone, constrOid);-- ObjectAddressSet(childAddr, ConstraintRelationId, constrOid);- recordDependencyOn(&childAddr, &parentAddr, DEPENDENCY_INTERNAL_AUTO);-- fkconstraint = makeNode(Constraint);- /* for now this is all we need */- fkconstraint->conname = pstrdup(NameStr(constrForm->conname));- fkconstraint->fk_upd_action = constrForm->confupdtype;- fkconstraint->fk_del_action = constrForm->confdeltype;- fkconstraint->deferrable = constrForm->condeferrable;- fkconstraint->initdeferred = constrForm->condeferred;-- createForeignKeyTriggers(partRel, constrForm->confrelid, fkconstraint,- constrOid, constrForm->conindid, false);-- if (cloned)- {- ClonedConstraint *newc;-- /*- * Feed back caller about the constraints we created, so that they- * can set up constraint verification.- */- newc = palloc(sizeof(ClonedConstraint));- newc->relid = RelationGetRelid(partRel);- newc->refrelid = constrForm->confrelid;- newc->conindid = constrForm->conindid;- newc->conid = constrOid;- newc->constraint = fkconstraint;-- *cloned = lappend(*cloned, newc);- }-- ReleaseSysCache(tuple);- }-- pfree(attmap);- list_free_deep(partFKs);-- /*- * If the partition is partitioned, recurse to handle any constraints that- * were cloned.- */- if (partRel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE &&- subclone != NIL)- {- PartitionDesc partdesc = RelationGetPartitionDesc(partRel);- int i;-- for (i = 0; i < partdesc->nparts; i++)- {- Relation childRel;-- childRel = heap_open(partdesc->oids[i], AccessExclusiveLock);- clone_fk_constraints(pg_constraint,- partRel,- childRel,- subclone,- cloned);- heap_close(childRel, NoLock); /* keep lock till commit */- }- }-}--/*
* Test whether given name is currently used as a constraint name
* for the given object (relation or domain).
*
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.cindex 843ed48aa7..31ca651213 100644--- a/src/backend/commands/tablecmds.c+++ b/src/backend/commands/tablecmds.c@@ -412,6 +412,10 @@ static ObjectAddress ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *
Relation rel, Constraint *fkconstraint, Oid parentConstr,
bool recurse, bool recursing,
LOCKMODE lockmode);
+static void CloneForeignKeyConstraints(Oid parentId, Oid relationId,+ List **cloned);+static void CloneFkReferencing(Relation pg_constraint, Relation parentRel,+ Relation partRel, List *clone, List **cloned);
static void ATExecDropConstraint(Relation rel, const char *constrName,
DropBehavior behavior,
bool recurse, bool recursing,
@@ -7631,6 +7635,308 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
}
/*
+ * CloneForeignKeyConstraints+ * Clone foreign keys from a partitioned table to a newly acquired+ * partition.+ *+ * relationId is a partition of parentId, so we can be certain that it has the+ * same columns with the same datatypes. The columns may be in different+ * order, though.+ *+ * The *cloned list is appended ClonedConstraint elements describing what was+ * created, for the purposes of validating the constraint in ALTER TABLE's+ * Phase 3.+ */+static void+CloneForeignKeyConstraints(Oid parentId, Oid relationId, List **cloned)+{+ Relation pg_constraint;+ Relation parentRel;+ Relation rel;+ ScanKeyData key;+ SysScanDesc scan;+ HeapTuple tuple;+ List *clone = NIL;++ parentRel = heap_open(parentId, NoLock); /* already got lock */+ /* see ATAddForeignKeyConstraint about lock level */+ rel = heap_open(relationId, AccessExclusiveLock);+ pg_constraint = heap_open(ConstraintRelationId, RowShareLock);++ /* Obtain the list of constraints to clone or attach */+ ScanKeyInit(&key,+ Anum_pg_constraint_conrelid, BTEqualStrategyNumber,+ F_OIDEQ, ObjectIdGetDatum(parentId));+ scan = systable_beginscan(pg_constraint, ConstraintRelidTypidNameIndexId, true,+ NULL, 1, &key);+ while ((tuple = systable_getnext(scan)) != NULL)+ {+ Oid oid = ((Form_pg_constraint) GETSTRUCT(tuple))->oid;++ clone = lappend_oid(clone, oid);+ }+ systable_endscan(scan);++ /* Do the actual work, recursing to partitions as needed */+ CloneFkReferencing(pg_constraint, parentRel, rel, clone, cloned);++ /* We're done. Clean up */+ heap_close(parentRel, NoLock);+ heap_close(rel, NoLock); /* keep lock till commit */+ heap_close(pg_constraint, RowShareLock);+}++/*+ * CloneFkReferencing+ * Recursive subroutine for CloneForeignKeyConstraints, referencing side+ *+ * Clone the given list of FK constraints when a partition is attached on the+ * referencing side of those constraints.+ *+ * When cloning foreign keys to a partition, it may happen that equivalent+ * constraints already exist in the partition for some of them. We can skip+ * creating a clone in that case, and instead just attach the existing+ * constraint to the one in the parent.+ *+ * This function recurses to partitions, if the new partition is partitioned;+ * of course, only do this for FKs that were actually cloned.+ */+static void+CloneFkReferencing(Relation pg_constraint, Relation parentRel,+ Relation partRel, List *clone, List **cloned)+{+ AttrNumber *attmap;+ List *partFKs;+ List *subclone = NIL;+ ListCell *cell;++ /*+ * The constraint key may differ, if the columns in the partition are+ * different. This map is used to convert them.+ */+ attmap = convert_tuples_by_name_map(RelationGetDescr(partRel),+ RelationGetDescr(parentRel),+ gettext_noop("could not convert row type"));++ partFKs = copyObject(RelationGetFKeyList(partRel));++ foreach(cell, clone)+ {+ Oid parentConstrOid = lfirst_oid(cell);+ Form_pg_constraint constrForm;+ HeapTuple tuple;+ int numfks;+ AttrNumber conkey[INDEX_MAX_KEYS];+ AttrNumber mapped_conkey[INDEX_MAX_KEYS];+ AttrNumber confkey[INDEX_MAX_KEYS];+ Oid conpfeqop[INDEX_MAX_KEYS];+ Oid conppeqop[INDEX_MAX_KEYS];+ Oid conffeqop[INDEX_MAX_KEYS];+ Constraint *fkconstraint;+ bool attach_it;+ Oid constrOid;+ ObjectAddress parentAddr,+ childAddr;+ ListCell *cell;+ int i;++ tuple = SearchSysCache1(CONSTROID, parentConstrOid);+ if (!tuple)+ elog(ERROR, "cache lookup failed for constraint %u",+ parentConstrOid);+ constrForm = (Form_pg_constraint) GETSTRUCT(tuple);++ /* only foreign keys */+ if (constrForm->contype != CONSTRAINT_FOREIGN)+ {+ ReleaseSysCache(tuple);+ continue;+ }++ ObjectAddressSet(parentAddr, ConstraintRelationId, parentConstrOid);++ DeconstructConstraintRow(tuple, &numfks, conkey, confkey,+ conpfeqop, conppeqop, conffeqop);+ for (i = 0; i < numfks; i++)+ mapped_conkey[i] = attmap[conkey[i] - 1];++ /*+ * Before creating a new constraint, see whether any existing FKs are+ * fit for the purpose. If one is, attach the parent constraint to it,+ * and don't clone anything. This way we avoid the expensive+ * verification step and don't end up with a duplicate FK. This also+ * means we don't consider this constraint when recursing to+ * partitions.+ */+ attach_it = false;+ foreach(cell, partFKs)+ {+ ForeignKeyCacheInfo *fk = lfirst_node(ForeignKeyCacheInfo, cell);+ Form_pg_constraint partConstr;+ HeapTuple partcontup;++ attach_it = true;++ /*+ * Do some quick & easy initial checks. If any of these fail, we+ * cannot use this constraint, but keep looking.+ */+ if (fk->confrelid != constrForm->confrelid || fk->nkeys != numfks)+ {+ attach_it = false;+ continue;+ }+ for (i = 0; i < numfks; i++)+ {+ if (fk->conkey[i] != mapped_conkey[i] ||+ fk->confkey[i] != confkey[i] ||+ fk->conpfeqop[i] != conpfeqop[i])+ {+ attach_it = false;+ break;+ }+ }+ if (!attach_it)+ continue;++ /*+ * Looks good so far; do some more extensive checks. Presumably+ * the check for 'convalidated' could be dropped, since we don't+ * really care about that, but let's be careful for now.+ */+ 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 (OidIsValid(partConstr->conparentid) ||+ !partConstr->convalidated ||+ partConstr->condeferrable != constrForm->condeferrable ||+ partConstr->condeferred != constrForm->condeferred ||+ partConstr->confupdtype != constrForm->confupdtype ||+ partConstr->confdeltype != constrForm->confdeltype ||+ partConstr->confmatchtype != constrForm->confmatchtype)+ {+ ReleaseSysCache(partcontup);+ attach_it = false;+ continue;+ }++ ReleaseSysCache(partcontup);++ /* looks good! Attach this constraint */+ ConstraintSetParentConstraint(fk->conoid, parentConstrOid);+ CommandCounterIncrement();+ attach_it = true;+ break;+ }++ /*+ * If we attached to an existing constraint, there is no need to+ * create a new one. In fact, there's no need to recurse for this+ * constraint to partitions, either.+ */+ if (attach_it)+ {+ ReleaseSysCache(tuple);+ continue;+ }++ constrOid =+ CreateConstraintEntry(NameStr(constrForm->conname),+ constrForm->connamespace,+ CONSTRAINT_FOREIGN,+ constrForm->condeferrable,+ constrForm->condeferred,+ constrForm->convalidated,+ parentConstrOid,+ RelationGetRelid(partRel),+ mapped_conkey,+ numfks,+ numfks,+ InvalidOid, /* not a domain constraint */+ constrForm->conindid, /* same index */+ constrForm->confrelid, /* same foreign rel */+ confkey,+ conpfeqop,+ conppeqop,+ conffeqop,+ numfks,+ constrForm->confupdtype,+ constrForm->confdeltype,+ constrForm->confmatchtype,+ NULL,+ NULL,+ NULL,+ false,+ 1, false, true);+ subclone = lappend_oid(subclone, constrOid);++ ObjectAddressSet(childAddr, ConstraintRelationId, constrOid);+ recordDependencyOn(&childAddr, &parentAddr, DEPENDENCY_INTERNAL_AUTO);++ fkconstraint = makeNode(Constraint);+ /* for now this is all we need */+ fkconstraint->conname = pstrdup(NameStr(constrForm->conname));+ fkconstraint->fk_upd_action = constrForm->confupdtype;+ fkconstraint->fk_del_action = constrForm->confdeltype;+ fkconstraint->deferrable = constrForm->condeferrable;+ fkconstraint->initdeferred = constrForm->condeferred;++ createForeignKeyTriggers(partRel, constrForm->confrelid, fkconstraint,+ constrOid, constrForm->conindid, false);++ if (cloned)+ {+ ClonedConstraint *newc;++ /*+ * Feed back caller about the constraints we created, so that they+ * can set up constraint verification.+ */+ newc = palloc(sizeof(ClonedConstraint));+ newc->relid = RelationGetRelid(partRel);+ newc->refrelid = constrForm->confrelid;+ newc->conindid = constrForm->conindid;+ newc->conid = constrOid;+ newc->constraint = fkconstraint;++ *cloned = lappend(*cloned, newc);+ }++ ReleaseSysCache(tuple);+ }++ pfree(attmap);+ list_free_deep(partFKs);++ /*+ * If the partition is partitioned, recurse to handle any constraints that+ * were cloned.+ */+ if (partRel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE &&+ subclone != NIL)+ {+ PartitionDesc partdesc = RelationGetPartitionDesc(partRel);+ int i;++ for (i = 0; i < partdesc->nparts; i++)+ {+ Relation childRel;++ childRel = heap_open(partdesc->oids[i], AccessExclusiveLock);+ CloneFkReferencing(pg_constraint,+ partRel,+ childRel,+ subclone,+ cloned);+ heap_close(childRel, NoLock); /* keep lock till commit */+ }+ }+}++/*
* ALTER TABLE ALTER CONSTRAINT
*
* Update the attributes of a constraint.
diff --git a/src/include/catalog/pg_constraint.h b/src/include/catalog/pg_constraint.hindex 8405fe2eea..8b2af36ba7 100644--- a/src/include/catalog/pg_constraint.h+++ b/src/include/catalog/pg_constraint.h@@ -226,9 +226,6 @@ extern Oid CreateConstraintEntry(const char *constraintName,
bool conNoInherit,
bool is_internal);
-extern void CloneForeignKeyConstraints(Oid parentId, Oid relationId,- List **cloned);-
extern void RemoveConstraintById(Oid conId);
extern void RenameConstraintById(Oid conId, const char *newname);
--
2.11.0
[text/x-diff] 0003-Rework-deleteObjectsInList-to-allow-objtype-specific.patch (2.2K, ../20181130191216.5xcxcsx3ascgqayv@alvherre.pgsql/4-0003-Rework-deleteObjectsInList-to-allow-objtype-specific.patch)
download | inline diff:
From b1024f44489f4d610e3658fa91642bf090692faf Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Wed, 28 Nov 2018 11:52:00 -0300
Subject: [PATCH 3/5] Rework deleteObjectsInList to allow objtype-specific
checks
This doesn't change any functionality yet.
---
src/backend/catalog/dependency.c | 41 ++++++++++++++++++++++++----------------
1 file changed, 25 insertions(+), 16 deletions(-)
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.cindex 7dfa3278a5..c7d3b8a654 100644--- a/src/backend/catalog/dependency.c+++ b/src/backend/catalog/dependency.c@@ -219,29 +219,38 @@ deleteObjectsInList(ObjectAddresses *targetObjects, Relation *depRel,
int i;
/*
- * Keep track of objects for event triggers, if necessary.+ * Invoke pre-deletion callbacks and keep track of objects for event+ * triggers, if necessary.
*/
- if (trackDroppedObjectsNeeded() && !(flags & PERFORM_DELETION_INTERNAL))+ for (i = 0; i < targetObjects->numrefs; i++)
{
- for (i = 0; i < targetObjects->numrefs; i++)+ const ObjectAddress *thisobj = &targetObjects->refs[i];+ Oid objectClass = getObjectClass(thisobj);++ if (trackDroppedObjectsNeeded() && !(flags & PERFORM_DELETION_INTERNAL))
{
- const ObjectAddress *thisobj = &targetObjects->refs[i];- const ObjectAddressExtra *extra = &targetObjects->extras[i];- bool original = false;- bool normal = false;-- if (extra->flags & DEPFLAG_ORIGINAL)- original = true;- if (extra->flags & DEPFLAG_NORMAL)- normal = true;- if (extra->flags & DEPFLAG_REVERSE)- normal = true;-- if (EventTriggerSupportsObjectClass(getObjectClass(thisobj)))+ if (EventTriggerSupportsObjectClass(objectClass))
{
+ bool original = false;+ bool normal = false;+ const ObjectAddressExtra *extra = &targetObjects->extras[i];++ if (extra->flags & DEPFLAG_ORIGINAL)+ original = true;+ if (extra->flags & DEPFLAG_NORMAL ||+ extra->flags & DEPFLAG_REVERSE)+ normal = true;+
EventTriggerSQLDropAddObject(thisobj, original, normal);
}
}
++ /* Invoke class-specific pre-deletion checks */+ switch (objectClass)+ {+ default:+ break;+ }
}
/*
--
2.11.0
[text/x-diff] 0004-Support-foreign-key-referencing-partitioned-tables.patch (69.7K, ../20181130191216.5xcxcsx3ascgqayv@alvherre.pgsql/5-0004-Support-foreign-key-referencing-partitioned-tables.patch)
download | inline diff:
From 44300eb0dfdf2221ed8fffd200fc7d6b958d090e Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Thu, 30 Aug 2018 06:01:38 -0300
Subject: [PATCH 4/5] Support foreign key referencing partitioned tables
---
src/backend/catalog/dependency.c | 3 +
src/backend/catalog/heap.c | 18 +
src/backend/catalog/pg_constraint.c | 15 +-
src/backend/commands/tablecmds.c | 813 +++++++++++++++++++++++++-----
src/backend/utils/adt/ri_triggers.c | 228 ++++++++-
src/backend/utils/adt/ruleutils.c | 18 +
src/include/catalog/heap.h | 2 +
src/include/commands/tablecmds.h | 6 +-
src/include/commands/trigger.h | 1 +
src/include/utils/ruleutils.h | 1 +
src/test/regress/expected/foreign_key.out | 168 +++++-
src/test/regress/sql/foreign_key.sql | 114 ++++-
12 files changed, 1201 insertions(+), 186 deletions(-)
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.cindex c7d3b8a654..35c648fb21 100644--- a/src/backend/catalog/dependency.c+++ b/src/backend/catalog/dependency.c@@ -248,6 +248,9 @@ deleteObjectsInList(ObjectAddresses *targetObjects, Relation *depRel,
/* Invoke class-specific pre-deletion checks */
switch (objectClass)
{
+ case OCLASS_CLASS:+ pre_drop_class_check(thisobj->objectId, thisobj->objectSubId);+ break;
default:
break;
}
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.cindex 11debaa780..55f9e58a39 100644--- a/src/backend/catalog/heap.c+++ b/src/backend/catalog/heap.c@@ -1788,6 +1788,24 @@ RemoveAttrDefaultById(Oid attrdefId)
}
/*
+ * XXX explain+ */+void+pre_drop_class_check(Oid relationId, Oid objectSubId)+{+ Relation relation;++ /* caller must hold strong lock already, if they're dropping */+ relation = relation_open(relationId, NoLock);++ if (relation->rd_rel->relkind == RELKIND_RELATION &&+ relation->rd_rel->relispartition)+ CheckNoForeignKeyRefs(relation, true);++ relation_close(relation, NoLock);+}++/*
* heap_drop_with_catalog - removes specified relation from catalogs
*
* Note that this routine is not responsible for dropping objects that are
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.cindex 2833e45cff..1c5017eccc 100644--- a/src/backend/catalog/pg_constraint.c+++ b/src/backend/catalog/pg_constraint.c@@ -303,10 +303,12 @@ CreateConstraintEntry(const char *constraintName,
if (OidIsValid(indexRelId) && constraintType == CONSTRAINT_FOREIGN)
{
/*
- * Register normal dependency on the unique index that supports a- * foreign-key constraint. (Note: for indexes associated with unique- * or primary-key constraints, the dependency runs the other way, and- * is not made here.)+ * Register dependency on the unique index that supports a foreign-key+ * constraint. (Note: for indexes associated with unique or+ * primary-key constraints, the dependency runs the other way, and is+ * not made here.) For standalone constraints, this is a normal+ * dependency; for a constraint that is implementation part of a+ * larger one, it's internal-auto.
*/
ObjectAddress relobject;
@@ -314,7 +316,10 @@ CreateConstraintEntry(const char *constraintName,
relobject.objectId = indexRelId;
relobject.objectSubId = 0;
- recordDependencyOn(&conobject, &relobject, DEPENDENCY_NORMAL);+ recordDependencyOn(&conobject, &relobject,+ OidIsValid(parentConstrId) ?+ DEPENDENCY_INTERNAL_AUTO :+ DEPENDENCY_NORMAL);
}
if (foreignNKeys > 0)
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.cindex 31ca651213..74109557ae 100644--- a/src/backend/commands/tablecmds.c+++ b/src/backend/commands/tablecmds.c@@ -412,10 +412,28 @@ static ObjectAddress ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *
Relation rel, Constraint *fkconstraint, Oid parentConstr,
bool recurse, bool recursing,
LOCKMODE lockmode);
+static ObjectAddress addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint,+ Relation rel, Relation pkrel, Oid indexOid, Oid parentConstraint,+ int numfks, int16 *pkattnum, int16 *fkattnum,+ Oid *pfeqoperators, Oid *ppeqoperators, Oid *ffeqoperators,+ bool old_check_ok);+static void addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint,+ Relation rel, Relation pkrel, Oid indexOid, Oid parentConstr, int numfks,+ int16 *pkattnum, int16 *fkattnum, Oid *pfeqoperators,+ Oid *ppeqoperators, Oid *ffeqoperators, bool old_check_ok,+ bool create_constraint);
static void CloneForeignKeyConstraints(Oid parentId, Oid relationId,
List **cloned);
+static void CloneFkReferenced(Relation parentRel, Relation partitionRel,+ Relation pg_constraint);
static void CloneFkReferencing(Relation pg_constraint, Relation parentRel,
Relation partRel, List *clone, List **cloned);
+static void createForeignKeyCheckTriggers(Oid myRelOid, Oid refRelOid,+ Constraint *fkconstraint, Oid constraintOid,+ Oid indexOid);+static void createForeignKeyActionTriggers(Relation rel, Oid refRelOid,+ Constraint *fkconstraint, Oid constraintOid,+ Oid indexOid);
static void ATExecDropConstraint(Relation rel, const char *constrName,
DropBehavior behavior,
bool recurse, bool recursing,
@@ -3407,7 +3425,8 @@ AlterTableGetLockLevel(List *cmds)
/*
* Removing constraints can affect SELECTs that have been
- * optimised assuming the constraint holds true.+ * optimised assuming the constraint holds true. See also+ * CloneFkReferenced.
*/
case AT_DropConstraint: /* as DROP INDEX */
case AT_DropNotNull: /* may change some SQL plans */
@@ -6987,11 +7006,7 @@ ATExecAddConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
break;
case CONSTR_FOREIGN:
-
/*
- * Note that we currently never recurse for FK constraints, so the- * "recurse" flag is silently ignored.- *
* Assign or validate constraint name
*/
if (newConstraint->conname)
@@ -7170,6 +7185,12 @@ ATAddCheckConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
* Subroutine for ATExecAddConstraint. Must already hold exclusive
* lock on the rel, and have done appropriate validity checks for it.
* We do permissions checks here, however.
+ *+ * When the referenced or referencing tables (or both) are partitioned,+ * multiple pg_constraint rows are required -- one for each partitioned table+ * and each partition on each side (fortunately, not one for every combination+ * thereof). We also need the appropriate triggers to be created on each leaf+ * partition.
*/
static ObjectAddress
ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
@@ -7189,7 +7210,6 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
int numfks,
numpks;
Oid indexOid;
- Oid constrOid;
bool old_check_ok;
ObjectAddress address;
ListCell *old_pfeqop_item = list_head(fkconstraint->old_conpfeqop);
@@ -7207,12 +7227,6 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
* Validity checks (permission checks wait till we have the column
* numbers)
*/
- if (pkrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)- ereport(ERROR,- (errcode(ERRCODE_WRONG_OBJECT_TYPE),- errmsg("cannot reference partitioned table \"%s\"",- RelationGetRelationName(pkrel))));-
if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
{
if (!recurse)
@@ -7230,7 +7244,8 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
errdetail("This feature is not yet supported on partitioned tables.")));
}
- if (pkrel->rd_rel->relkind != RELKIND_RELATION)+ if (pkrel->rd_rel->relkind != RELKIND_RELATION &&+ pkrel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("referenced relation \"%s\" is not a table",
@@ -7440,8 +7455,9 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
if (!(OidIsValid(pfeqop) && OidIsValid(ffeqop)))
ereport(ERROR,
(errcode(ERRCODE_DATATYPE_MISMATCH),
- errmsg("foreign key constraint \"%s\" "- "cannot be implemented",+ errmsg(fkconstraint->conname ?+ "foreign key constraint \"%s\" cannot be implemented" :+ "foreign key constraint cannot be implemented",
fkconstraint->conname),
errdetail("Key columns \"%s\" and \"%s\" "
"are of incompatible types: %s and %s.",
@@ -7529,15 +7545,160 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
}
/*
+ * Create all the constraint and trigger objects, recursing to partitions+ * as necessary. First handle the referenced side.+ */+ address = addFkRecurseReferenced(wqueue, fkconstraint, rel, pkrel,+ indexOid,+ InvalidOid, /* no parent constraint */+ numfks,+ pkattnum,+ fkattnum,+ pfeqoperators,+ ppeqoperators,+ ffeqoperators,+ old_check_ok);++ /*+ * Now handle the referencing side. We don't need the topmost constraint+ * on this side, because the constraint we created above fills that role;+ * if this recurses for any partitions then more constraints are added.+ */+ addFkRecurseReferencing(wqueue, fkconstraint, rel, pkrel,+ indexOid,+ address.objectId,+ numfks,+ pkattnum,+ fkattnum,+ pfeqoperators,+ ppeqoperators,+ ffeqoperators,+ old_check_ok,+ false);++ /*+ * Close pk table, but keep lock until we've committed.+ */+ heap_close(pkrel, NoLock);++ return address;+}++/*+ * Return the OID of the index, in the given partition, that is a child of the+ * given index or InvalidOid if there isn't one.+ *+ * XXX maybe this should be in catalog/partition.c? Not for the moment only+ * because all the callers are here ...+ */+static Oid+index_get_partition(Relation partition, Oid indexId)+{+ List *idxlist = RelationGetIndexList(partition);+ ListCell *l;++ foreach(l, idxlist)+ {+ Oid partIdx = lfirst_oid(l);+ HeapTuple tup;+ Form_pg_class classForm;+ bool ispartition;++ tup = SearchSysCache1(RELOID, ObjectIdGetDatum(partIdx));+ if (!tup)+ elog(ERROR, "cache lookup failed for relation %u", partIdx);+ classForm = (Form_pg_class) GETSTRUCT(tup);+ ispartition = classForm->relispartition;+ ReleaseSysCache(tup);+ if (!ispartition)+ continue;+ if (get_partition_parent(lfirst_oid(l)) == indexId)+ {+ list_free(idxlist);+ return partIdx;+ }+ }++ return InvalidOid;+}++/*+ * addFkRecurseReferenced+ * recursive subroutine for ATAddForeignKeyConstraint, referenced side+ *+ * Create pg_constraint rows for the referenced side of the constraint,+ * referencing the parent of the referencing side; also create action triggers+ * on leaf partitions. If the table is partitioned, recurse to handle each+ * partition.+ *+ * On constraint names: It's not possible in general to give all the cascaded+ * constraints the same name, so we don't try. (Also, even when it *is*+ * possible, it goes counter to the SQL-standard rule that constraint names+ * must be unique within a schema.) Therefore we apply the given name to the+ * top-most constraint, and use generated names when cascading to partitions+ *+ * wqueue is the ALTER TABLE work queue; can be NULL when not running as part+ * of an ALTER TABLE sequence.+ * fkconstraint is the constraint being added.+ * rel is the root referencing relation.+ * pkrel is the referenced relation; might be a partition, if recursing.+ * indexOid is the OID of the index (on pkrel) implementing this constraint.+ * parentConstraint is the OID of a parent constraint; InvalidOid if this is a+ * top-level constraint.+ * numfks is the number of columns in the foreign key+ * pkattnum is the attnum array of referenced attributes.+ * fkattnum is the attnum array of referencing attributes.+ * pf/pp/ffeqoperators are OID array of operators between columns.+ * old_check_ok signals that this constraint replaces an existing one that+ * was already validated (thus this one doesn't need validation).+ */+static ObjectAddress+addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel,+ Relation pkrel, Oid indexOid, Oid parentConstraint,+ int numfks,+ int16 *pkattnum, int16 *fkattnum, Oid *pfeqoperators,+ Oid *ppeqoperators, Oid *ffeqoperators, bool old_check_ok)+{+ ObjectAddress address;+ Oid constrOid;+ char *conname;++ /*+ * Verify relkind for each referenced partition. At the top level, this+ * is redundant with a previous check, but we need it when recursing.+ */+ if (pkrel->rd_rel->relkind != RELKIND_RELATION &&+ pkrel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)+ ereport(ERROR,+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),+ errmsg("referenced relation \"%s\" is not a table",+ RelationGetRelationName(pkrel))));++ /*+ * Caller supplies us with a constraint name; however, it may be+ * used in this partition, so come up with a different one in that+ * case.+ */+ if (ConstraintNameIsUsed(CONSTRAINT_RELATION,+ RelationGetRelid(rel),+ fkconstraint->conname))+ conname = ChooseConstraintName(RelationGetRelationName(rel),+ strVal(linitial(fkconstraint->fk_attrs)),+ "fkey",+ RelationGetNamespace(rel), NIL);+ else+ conname = fkconstraint->conname;++ /*
* Record the FK constraint in pg_constraint.
*/
- constrOid = CreateConstraintEntry(fkconstraint->conname,+ constrOid = CreateConstraintEntry(conname,
RelationGetNamespace(rel),
CONSTRAINT_FOREIGN,
fkconstraint->deferrable,
fkconstraint->initdeferred,
fkconstraint->initially_valid,
- parentConstr,+ parentConstraint,
RelationGetRelid(rel),
fkattnum,
numfks,
@@ -7549,7 +7710,7 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
pfeqoperators,
ppeqoperators,
ffeqoperators,
- numpks,+ numfks,
fkconstraint->fk_upd_action,
fkconstraint->fk_del_action,
fkconstraint->fk_matchtype,
@@ -7562,79 +7723,262 @@ ATAddForeignKeyConstraint(List **wqueue, AlteredTableInfo *tab, Relation rel,
false); /* is_internal */
ObjectAddressSet(address, ConstraintRelationId, constrOid);
- /*- * Create the triggers that will enforce the constraint. We only want the- * action triggers to appear for the parent partitioned relation, even- * though the constraints also exist below.- */- createForeignKeyTriggers(rel, RelationGetRelid(pkrel), fkconstraint,- constrOid, indexOid, !recursing);+ /* make new constraint visible, in case we add more */+ CommandCounterIncrement();
/*
- * Tell Phase 3 to check that the constraint is satisfied by existing- * rows. We can skip this during table creation, when requested explicitly- * by specifying NOT VALID in an ADD FOREIGN KEY command, and when we're- * recreating a constraint following a SET DATA TYPE operation that did- * not impugn its validity.+ * If the referenced table is a plain relation, create the action triggers+ * that enforce the constraint.
*/
- if (!old_check_ok && !fkconstraint->skip_validation)+ if (pkrel->rd_rel->relkind == RELKIND_RELATION)
{
- NewConstraint *newcon;-- newcon = (NewConstraint *) palloc0(sizeof(NewConstraint));- newcon->name = fkconstraint->conname;- newcon->contype = CONSTR_FOREIGN;- newcon->refrelid = RelationGetRelid(pkrel);- newcon->refindid = indexOid;- newcon->conid = constrOid;- newcon->qual = (Node *) fkconstraint;-- tab->constraints = lappend(tab->constraints, newcon);+ createForeignKeyActionTriggers(rel, RelationGetRelid(pkrel),+ fkconstraint,+ constrOid, indexOid);
}
/*
- * When called on a partitioned table, recurse to create the constraint on- * the partitions also.+ * If the referenced table is partitioned, recurse on ourselves to handle+ * each partition. We need one pg_constraint row created for each+ * partition in addition to the pg_constraint row for the parent table.
*/
- if (recurse && rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)+ if (pkrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
{
- PartitionDesc partdesc;+ PartitionDesc pd = RelationGetPartitionDesc(pkrel);+ int i;- partdesc = RelationGetPartitionDesc(rel);-- for (i = 0; i < partdesc->nparts; i++)+ for (i = 0; i < pd->nparts; i++)
{
- Oid partitionId = partdesc->oids[i];- Relation partition = heap_open(partitionId, lockmode);- AlteredTableInfo *childtab;- ObjectAddress childAddr;+ Relation partRel;+ AttrNumber *map;+ AttrNumber *mapped_pkattnum;+ Oid partIndexId;- CheckTableNotInUse(partition, "ALTER TABLE");+ partRel = heap_open(pd->oids[i], ShareRowExclusiveLock);- /* Find or create work queue entry for this table */- childtab = ATGetQueueEntry(wqueue, partition);+ /*+ * Map the attribute numbers in the referenced side of the FK+ * definition to match the partition's column layout.+ */+ map = convert_tuples_by_name_map_if_req(RelationGetDescr(partRel),+ RelationGetDescr(pkrel),+ gettext_noop("could not convert row type"));+ if (map)+ {+ int j;- childAddr =- ATAddForeignKeyConstraint(wqueue, childtab, partition,- fkconstraint, constrOid,- recurse, true, lockmode);+ mapped_pkattnum = palloc(sizeof(AttrNumber) * numfks);+ for (j = 0; j < numfks; j++)+ mapped_pkattnum[j] = map[pkattnum[j] - 1];+ }+ else+ mapped_pkattnum = pkattnum;- /* Record this constraint as dependent on the parent one */- recordDependencyOn(&childAddr, &address, DEPENDENCY_INTERNAL_AUTO);+ /* do the deed */+ partIndexId = index_get_partition(partRel, indexOid);+ if (!OidIsValid(partIndexId))+ elog(ERROR, "index for %u not found in partition %s",+ indexOid, RelationGetRelationName(partRel));+ addFkRecurseReferenced(wqueue, fkconstraint, rel, partRel,+ partIndexId, constrOid, numfks,+ mapped_pkattnum, fkattnum,+ pfeqoperators, ppeqoperators, ffeqoperators,+ old_check_ok);- heap_close(partition, NoLock);+ /* Done -- clean up (but keep the lock) */+ heap_close(partRel, NoLock);+ if (map)+ {+ pfree(mapped_pkattnum);+ pfree(map);+ }
}
}
- /*- * Close pk table, but keep lock until we've committed.- */- heap_close(pkrel, NoLock);-
return address;
}
/*
+ * addFkRecurseReferencing+ * recursive subroutine for ATAddForeignKeyConstraint, referencing side+ *+ * wqueue is the ALTER TABLE work queue; can be NULL when not running as part+ * of an ALTER TABLE sequence.+ * fkconstraint is the constraint being added.+ * rel is the referencing relation; might be a partition, if recursing.+ * pkrel is the root referenced relation.+ * indexOid is the OID of the index (on pkrel) implementing this constraint.+ * parentConstr is the OID of the parent constraint (there is always one).+ * numfks is the number of columns in the foreign key+ * pkattnum is the attnum array of referenced attributes.+ * fkattnum is the attnum array of referencing attributes.+ * pf/pp/ffeqoperators are OID array of operators between columns.+ * old_check_ok signals that this constraint replaces an existing one that+ * was already validated (thus this one doesn't need validation).+ * create_constraint indicates whether to create a constraint; if false, the+ * given parentConstr is the origin of all others.+ *+ * Note we never try to use the constraint name assigned in fkconstraint.+ */+static void+addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel,+ Relation pkrel, Oid indexOid, Oid parentConstr,+ int numfks, int16 *pkattnum, int16 *fkattnum,+ Oid *pfeqoperators, Oid *ppeqoperators, Oid *ffeqoperators,+ bool old_check_ok, bool create_constraint)+{+ Oid constrOid; /* uninitialized */+ char *conname;++ AssertArg(OidIsValid(parentConstr));++ /*+ * Verify relkind for each referencing partition. At the top level, this+ * is redundant with a previous check, but we need it when recursing.+ */+ if (pkrel->rd_rel->relkind != RELKIND_RELATION &&+ pkrel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)+ ereport(ERROR,+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),+ errmsg("referenced relation \"%s\" is not a table",+ RelationGetRelationName(pkrel))));++ if (!create_constraint)+ constrOid = parentConstr;+ else+ {+ /* Figure out a good constraint name to use */+ conname = ChooseConstraintName(RelationGetRelationName(rel),+ strVal(linitial(fkconstraint->fk_attrs)),+ "fkey",+ RelationGetNamespace(rel), NIL);++ /*+ * Record the FK constraint in pg_constraint.+ */+ constrOid = CreateConstraintEntry(conname,+ RelationGetNamespace(rel),+ CONSTRAINT_FOREIGN,+ fkconstraint->deferrable,+ fkconstraint->initdeferred,+ fkconstraint->initially_valid,+ parentConstr,+ RelationGetRelid(rel),+ fkattnum,+ numfks,+ numfks,+ InvalidOid,+ indexOid,+ RelationGetRelid(pkrel),+ pkattnum,+ pfeqoperators,+ ppeqoperators,+ ffeqoperators,+ numfks,+ fkconstraint->fk_upd_action,+ fkconstraint->fk_del_action,+ fkconstraint->fk_matchtype,+ NULL, /* no exclusion constraint */+ NULL, /* no check constraint */+ NULL,+ true, /* islocal */+ 0, /* inhcount */+ true, /* isnoinherit */+ false); /* is_internal */+ /* make constraint visible */+ CommandCounterIncrement();+ }++ /*+ * If the referencing relation is a plain table, add the check triggers to+ * it and, if necessary, schedule it to be checked in Phase 3.+ *+ * If the relation is partitioned, drill down to do it to its partitions.+ */+ if (rel->rd_rel->relkind == RELKIND_RELATION)+ {+ createForeignKeyCheckTriggers(RelationGetRelid(rel),+ RelationGetRelid(pkrel),+ fkconstraint,+ constrOid,+ indexOid);++ /*+ * Tell Phase 3 to check that the constraint is satisfied by existing+ * rows. We can skip this during table creation, when requested+ * explicitly by specifying NOT VALID in an ADD FOREIGN KEY command,+ * and when we're recreating a constraint following a SET DATA TYPE+ * operation that did not impugn its validity.+ */+ if (!old_check_ok && !fkconstraint->skip_validation)+ {+ NewConstraint *newcon;+ AlteredTableInfo *tab;++ tab = ATGetQueueEntry(wqueue, rel);++ newcon = (NewConstraint *) palloc0(sizeof(NewConstraint));+ newcon->name = fkconstraint->conname;+ newcon->contype = CONSTR_FOREIGN;+ newcon->refrelid = RelationGetRelid(pkrel);+ newcon->refindid = indexOid;+ newcon->conid = constrOid;+ newcon->qual = (Node *) fkconstraint;++ tab->constraints = lappend(tab->constraints, newcon);+ }+ }+ else if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)+ {+ PartitionDesc pd;+ int i;++ pd = RelationGetPartitionDesc(rel);+ for (i = 0; i < pd->nparts; i++)+ {+ Relation partRel;+ AttrNumber *map;+ AttrNumber *mapped_fkattnum;++ partRel = heap_open(pd->oids[i], ShareRowExclusiveLock);++ /*+ * Map the attribute numbers in the referencing side of the FK+ * definition to match the partition's column layout.+ */+ map = convert_tuples_by_name_map_if_req(RelationGetDescr(partRel),+ RelationGetDescr(rel),+ gettext_noop("could not convert row type"));+ if (map)+ {+ int j;++ mapped_fkattnum = palloc(sizeof(AttrNumber) * numfks);+ for (j = 0; j < numfks; j++)+ mapped_fkattnum[j] = map[fkattnum[j] - 1];+ }+ else+ mapped_fkattnum = fkattnum;++ /* down the rabbit hole */+ addFkRecurseReferencing(wqueue, fkconstraint, partRel, pkrel,+ indexOid, constrOid, numfks, pkattnum,+ mapped_fkattnum,+ pfeqoperators, ppeqoperators, ffeqoperators,+ old_check_ok, true);++ /* Done -- clean up (but keep the lock) */+ heap_close(partRel, NoLock);+ if (map)+ {+ pfree(mapped_fkattnum);+ pfree(map);+ }+ }+ }+}++/*
* CloneForeignKeyConstraints
* Clone foreign keys from a partitioned table to a newly acquired
* partition.
@@ -7663,7 +8007,14 @@ CloneForeignKeyConstraints(Oid parentId, Oid relationId, List **cloned)
rel = heap_open(relationId, AccessExclusiveLock);
pg_constraint = heap_open(ConstraintRelationId, RowShareLock);
- /* Obtain the list of constraints to clone or attach */+ /*+ * Clone constraints where the parent is in the referenced side.+ */+ CloneFkReferenced(parentRel, rel, pg_constraint);++ /*+ * Now search for constraints where the parent is in the referencing side.+ */
ScanKeyInit(&key,
Anum_pg_constraint_conrelid, BTEqualStrategyNumber,
F_OIDEQ, ObjectIdGetDatum(parentId));
@@ -7680,13 +8031,160 @@ CloneForeignKeyConstraints(Oid parentId, Oid relationId, List **cloned)
/* Do the actual work, recursing to partitions as needed */
CloneFkReferencing(pg_constraint, parentRel, rel, clone, cloned);
- /* We're done. Clean up */+ list_free(clone);++ /* We're done. Clean up, keeping locks till commit */
heap_close(parentRel, NoLock);
- heap_close(rel, NoLock); /* keep lock till commit */+ heap_close(rel, NoLock);
heap_close(pg_constraint, RowShareLock);
}
/*
+ * CloneFkReferenced+ * Subroutine for CloneForeignKeyConstraints, referenced side+ *+ * Clone the FKs that reference the parent relation. Used when partitionRel+ * is created/attached. (Recursion to partitions is effected by callee+ * addFkRecurseReferenced, so this routine is not itself recursive.)+ */+static void+CloneFkReferenced(Relation parentRel, Relation partitionRel,+ Relation pg_constraint)+{+ AttrNumber *attmap;+ ListCell *cell;+ SysScanDesc scan;+ ScanKeyData key[2];+ HeapTuple tuple;+ List *clone = NIL;++ /*+ * Search for any constraints where this partition is in the referenced+ * side. However, we must ignore any constraint whose parent constraint+ * is also going to be cloned, to avoid duplicates. So do it in two+ * steps: first construct the list of constraints to clone, then go over+ * that list cloning those whose parents are not in the list. (We must+ * not rely on the parent being seen first, since catalog order could+ * return children first.)+ */+ attmap = convert_tuples_by_name_map(RelationGetDescr(partitionRel),+ RelationGetDescr(parentRel),+ gettext_noop("could not convert row type"));+ ScanKeyInit(&key[0],+ Anum_pg_constraint_confrelid, BTEqualStrategyNumber,+ F_OIDEQ, ObjectIdGetDatum(RelationGetRelid(parentRel)));+ ScanKeyInit(&key[1],+ Anum_pg_constraint_contype, BTEqualStrategyNumber,+ F_CHAREQ, CharGetDatum(CONSTRAINT_FOREIGN));+ /* This is a seqscan, as we don't have a usable index ... */+ scan = systable_beginscan(pg_constraint, InvalidOid, true,+ NULL, 2, key);+ while ((tuple = systable_getnext(scan)) != NULL)+ {+ Form_pg_constraint constrForm = (Form_pg_constraint) GETSTRUCT(tuple);++ /* ignore this constraint if the parent is already on the list */+ if (list_member_oid(clone, constrForm->conparentid))+ continue;++ clone = lappend_oid(clone, constrForm->oid);+ }+ systable_endscan(scan);++ foreach(cell, clone)+ {+ Oid constrOid = lfirst_oid(cell);+ Form_pg_constraint constrForm;+ Form_pg_attribute att;+ Relation fkRel;+ Oid indexOid;+ Oid partIndexId;+ int numfks;+ AttrNumber conkey[INDEX_MAX_KEYS];+ AttrNumber mapped_confkey[INDEX_MAX_KEYS];+ AttrNumber confkey[INDEX_MAX_KEYS];+ Oid conpfeqop[INDEX_MAX_KEYS];+ Oid conppeqop[INDEX_MAX_KEYS];+ Oid conffeqop[INDEX_MAX_KEYS];+ Constraint *fkconstraint;++ tuple = SearchSysCache1(CONSTROID, constrOid);+ if (!HeapTupleIsValid(tuple))+ elog(ERROR, "cache lookup failed for constraint %u", constrOid);+ constrForm = (Form_pg_constraint) GETSTRUCT(tuple);++ /* skip children whose parents are going to be cloned, as above */+ if (list_member_oid(clone, constrForm->conparentid))+ {+ ReleaseSysCache(tuple);+ continue;+ }++ /*+ * Because we're only expanding the key space at the referenced side,+ * we don't need to prevent any operation in the referencing table, so+ * AccessShareLock suffices (assumes that dropping the constraint+ * acquires AEL).+ */+ fkRel = heap_open(constrForm->conrelid, AccessShareLock);++ indexOid = constrForm->conindid;+ DeconstructConstraintRow(tuple,+ &numfks,+ conkey,+ confkey,+ conpfeqop,+ conppeqop,+ conffeqop);+ for (int i = 0; i < numfks; i++)+ mapped_confkey[i] = attmap[confkey[i] - 1];++ fkconstraint = makeNode(Constraint);+ /* for now this is all we need */+ fkconstraint->conname = NameStr(constrForm->conname);+ fkconstraint->fk_upd_action = constrForm->confupdtype;+ fkconstraint->fk_del_action = constrForm->confdeltype;+ fkconstraint->deferrable = constrForm->condeferrable;+ fkconstraint->initdeferred = constrForm->condeferred;+ fkconstraint->initially_valid = true;+ fkconstraint->fk_matchtype = constrForm->confmatchtype;++ /*+ * This is a bit grotty: we only need the first column name, which is+ * used to generate the constraint name.+ */+ att = TupleDescAttr(RelationGetDescr(partitionRel), conkey[0] - 1);+ fkconstraint->fk_attrs = list_make1(makeString(NameStr(att->attname)));++ /*+ * Add the new foreign key constraint pointing to the new partition.+ * Because this new partition appears in the referenced side of the+ * constraint, we don't need to set up for Phase 3 check.+ */+ partIndexId = index_get_partition(partitionRel, indexOid);+ if (!OidIsValid(partIndexId))+ elog(ERROR, "index for %u not found in partition %s",+ indexOid, RelationGetRelationName(partitionRel));+ addFkRecurseReferenced(NULL,+ fkconstraint,+ fkRel,+ partitionRel,+ partIndexId,+ constrOid,+ numfks,+ mapped_confkey,+ conkey,+ conpfeqop,+ conppeqop,+ conffeqop,+ true);++ heap_close(fkRel, NoLock);+ ReleaseSysCache(tuple);+ }+}++/*
* CloneFkReferencing
* Recursive subroutine for CloneForeignKeyConstraints, referencing side
*
@@ -7752,18 +8250,24 @@ CloneFkReferencing(Relation pg_constraint, Relation parentRel,
ReleaseSysCache(tuple);
continue;
}
+ if (list_member_oid(clone, constrForm->conparentid))+ {+ ReleaseSysCache(tuple);+ continue;+ }
ObjectAddressSet(parentAddr, ConstraintRelationId, parentConstrOid);
DeconstructConstraintRow(tuple, &numfks, conkey, confkey,
conpfeqop, conppeqop, conffeqop);
+
for (i = 0; i < numfks; i++)
mapped_conkey[i] = attmap[conkey[i] - 1];
/*
* Before creating a new constraint, see whether any existing FKs are
- * fit for the purpose. If one is, attach the parent constraint to it,- * and don't clone anything. This way we avoid the expensive+ * fit for the purpose. If one is, attach the parent constraint to+ * it, and don't clone anything. This way we avoid the expensive
* verification step and don't end up with a duplicate FK. This also
* means we don't consider this constraint when recursing to
* partitions.
@@ -7884,8 +8388,12 @@ CloneFkReferencing(Relation pg_constraint, Relation parentRel,
fkconstraint->deferrable = constrForm->condeferrable;
fkconstraint->initdeferred = constrForm->condeferred;
- createForeignKeyTriggers(partRel, constrForm->confrelid, fkconstraint,- constrOid, constrForm->conindid, false);+ /* If this is a plain relation, create the check triggers */+ if (partRel->rd_rel->relkind == RELKIND_RELATION)+ createForeignKeyCheckTriggers(RelationGetRelid(partRel),+ constrForm->confrelid,+ fkconstraint, constrOid,+ constrForm->conindid);
if (cloned)
{
@@ -7921,6 +8429,9 @@ CloneFkReferencing(Relation pg_constraint, Relation parentRel,
PartitionDesc partdesc = RelationGetPartitionDesc(partRel);
int i;
+ /* make previously created constraints visible */+ CommandCounterIncrement();+
for (i = 0; i < partdesc->nparts; i++)
{
Relation childRel;
@@ -8968,37 +9479,6 @@ createForeignKeyCheckTriggers(Oid myRelOid, Oid refRelOid,
}
/*
- * Create the triggers that implement an FK constraint.- *- * NB: if you change any trigger properties here, see also- * ATExecAlterConstraint.- */-void-createForeignKeyTriggers(Relation rel, Oid refRelOid, Constraint *fkconstraint,- Oid constraintOid, Oid indexOid, bool create_action)-{- /*- * For the referenced side, create action triggers, if requested. (If the- * referencing side is partitioned, there is still only one trigger, which- * runs on the referenced side and points to the top of the referencing- * hierarchy.)- */- if (create_action)- createForeignKeyActionTriggers(rel, refRelOid, fkconstraint, constraintOid,- indexOid);-- /*- * For the referencing side, create the check triggers. We only need- * these on the partitions.- */- if (rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE)- createForeignKeyCheckTriggers(RelationGetRelid(rel), refRelOid,- fkconstraint, constraintOid, indexOid);-- CommandCounterIncrement();-}--/*
* ALTER TABLE DROP CONSTRAINT
*
* Like DROP COLUMN, we can't use the normal ALTER TABLE recursion mechanism.
@@ -14848,6 +15328,9 @@ ATExecDetachPartition(Relation rel, RangeVar *name)
partRel = heap_openrv(name, AccessShareLock);
+ /* Ensure that foreign keys still hold after this detach */+ CheckNoForeignKeyRefs(partRel, false);+
/* All inheritance related checks are performed within the function */
RemoveInheritance(partRel, rel);
@@ -15164,36 +15647,17 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name)
static void
refuseDupeIndexAttach(Relation parentIdx, Relation partIdx, Relation partitionTbl)
{
- Relation pg_inherits;- ScanKeyData key;- HeapTuple tuple;- SysScanDesc scan;+ Oid existingIdx;- pg_inherits = heap_open(InheritsRelationId, AccessShareLock);- ScanKeyInit(&key, Anum_pg_inherits_inhparent,- BTEqualStrategyNumber, F_OIDEQ,- ObjectIdGetDatum(RelationGetRelid(parentIdx)));- scan = systable_beginscan(pg_inherits, InheritsParentIndexId, true,- NULL, 1, &key);- while (HeapTupleIsValid(tuple = systable_getnext(scan)))- {- Form_pg_inherits inhForm;- Oid tab;-- inhForm = (Form_pg_inherits) GETSTRUCT(tuple);- tab = IndexGetRelation(inhForm->inhrelid, false);- if (tab == RelationGetRelid(partitionTbl))- ereport(ERROR,- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),- errmsg("cannot attach index \"%s\" as a partition of index \"%s\"",- RelationGetRelationName(partIdx),- RelationGetRelationName(parentIdx)),- errdetail("Another index is already attached for partition \"%s\".",- RelationGetRelationName(partitionTbl))));- }-- systable_endscan(scan);- heap_close(pg_inherits, AccessShareLock);+ existingIdx = index_get_partition(partitionTbl, RelationGetRelid(parentIdx));+ if (OidIsValid(existingIdx))+ ereport(ERROR,+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),+ errmsg("cannot attach index \"%s\" as a partition of index \"%s\"",+ RelationGetRelationName(partIdx),+ RelationGetRelationName(parentIdx)),+ errdetail("Another index is already attached for partition \"%s\".",+ RelationGetRelationName(partitionTbl))));
}
/*
@@ -15324,3 +15788,82 @@ update_relispartition(Relation classRel, Oid relationId, bool newval)
if (opened)
heap_close(classRel, RowExclusiveLock);
}
++/*+ * During an operation that removes a partition from a partitioned table,+ * verify that any foreign keys pointing to the partitioned table would not+ * become invalid. (Used for ALTER TABLE ... DETACH as well as DROP).+ * An error raised if any referenced values exist.+ */+void+CheckNoForeignKeyRefs(Relation partition, bool isDrop)+{+ Relation pg_constraint;+ HeapTuple tuple;+ SysScanDesc scan;+ ScanKeyData key[2];++ /*+ * Quick exit if there are no indexes.+ */+ if (RelationGetIndexList(partition) == NIL)+ return;++ /*+ * In the DROP case, we can skip this check when this is a partitioned+ * partition, because its partitions will go through this also, and we'd+ * run the check twice uselessly.+ *+ * In the DETACH case, this is only called for the top-level relation,+ * so we must run it nevertheless.+ */+ if (isDrop && partition->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)+ return;++ /* Search for constraints referencing this table */+ pg_constraint = heap_open(ConstraintRelationId, AccessShareLock);+ ScanKeyInit(&key[0],+ Anum_pg_constraint_confrelid, BTEqualStrategyNumber,+ F_OIDEQ, ObjectIdGetDatum(RelationGetRelid(partition)));+ ScanKeyInit(&key[1],+ Anum_pg_constraint_contype, BTEqualStrategyNumber,+ F_CHAREQ, CharGetDatum(CONSTRAINT_FOREIGN));++ /* XXX This is a seqscan, as we don't have a usable index */+ scan = systable_beginscan(pg_constraint, InvalidOid, true,+ NULL, 2, key);+ while ((tuple = systable_getnext(scan)) != NULL)+ {+ Form_pg_constraint constrForm = (Form_pg_constraint) GETSTRUCT(tuple);+ Trigger trig;+ Relation rel;++ /*+ * We only need to process constraints that are part of larger ones.+ */+ if (!OidIsValid(constrForm->conparentid))+ continue;++ /* prevent data changes into the referencing table until commit */+ rel = heap_open(constrForm->conrelid, ShareLock);++ MemSet(&trig, 0, sizeof(trig));+ trig.tgoid = InvalidOid;+ trig.tgname = NameStr(constrForm->conname);+ trig.tgenabled = TRIGGER_FIRES_ON_ORIGIN;+ trig.tgisinternal = true;+ trig.tgconstrrelid = RelationGetRelid(partition);+ trig.tgconstrindid = constrForm->conindid;+ trig.tgconstraint = constrForm->oid;+ trig.tgdeferrable = false;+ trig.tginitdeferred = false;+ /* we needn't fill in remaining fields */++ RI_Final_Check(&trig, rel, partition);++ heap_close(rel, NoLock);+ }++ systable_endscan(scan);+ heap_close(pg_constraint, AccessShareLock);+}diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.cindex b9aef07c1a..d7e03adeae 100644--- a/src/backend/utils/adt/ri_triggers.c+++ b/src/backend/utils/adt/ri_triggers.c@@ -54,6 +54,7 @@
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/rls.h"
+#include "utils/ruleutils.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/tqual.h"
@@ -239,7 +240,7 @@ static void ri_ExtractValues(Relation rel, HeapTuple tup,
static void ri_ReportViolation(const RI_ConstraintInfo *riinfo,
Relation pk_rel, Relation fk_rel,
HeapTuple violator, TupleDesc tupdesc,
- int queryno) pg_attribute_noreturn();+ int queryno, bool partgone) pg_attribute_noreturn();
/* ----------
@@ -397,18 +398,22 @@ RI_FKey_check(TriggerData *trigdata)
char paramname[16];
const char *querysep;
Oid queryoids[RI_MAX_NUMKEYS];
+ const char *pk_only;
/* ----------
* The query string built is
- * SELECT 1 FROM ONLY <pktable> x WHERE pkatt1 = $1 [AND ...]+ * SELECT 1 FROM [ONLY] <pktable> x WHERE pkatt1 = $1 [AND ...]
* FOR KEY SHARE OF x
* The type id's for the $ parameters are those of the
* corresponding FK attributes.
* ----------
*/
initStringInfo(&querybuf);
+ pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?+ "" : "ONLY ";
quoteRelationName(pkrelname, pk_rel);
- appendStringInfo(&querybuf, "SELECT 1 FROM ONLY %s x", pkrelname);+ appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",+ pk_only, pkrelname);
querysep = "WHERE";
for (i = 0; i < riinfo->nkeys; i++)
{
@@ -532,19 +537,23 @@ ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
char attname[MAX_QUOTED_NAME_LEN];
char paramname[16];
const char *querysep;
+ const char *pk_only;
Oid queryoids[RI_MAX_NUMKEYS];
/* ----------
* The query string built is
- * SELECT 1 FROM ONLY <pktable> x WHERE pkatt1 = $1 [AND ...]+ * SELECT 1 FROM [ONLY] <pktable> x WHERE pkatt1 = $1 [AND ...]
* FOR KEY SHARE OF x
* The type id's for the $ parameters are those of the
* PK attributes themselves.
* ----------
*/
initStringInfo(&querybuf);
+ pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?+ "" : "ONLY ";
quoteRelationName(pkrelname, pk_rel);
- appendStringInfo(&querybuf, "SELECT 1 FROM ONLY %s x", pkrelname);+ appendStringInfo(&querybuf, "SELECT 1 FROM %s%s x",+ pk_only, pkrelname);
querysep = "WHERE";
for (i = 0; i < riinfo->nkeys; i++)
{
@@ -1710,6 +1719,7 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
RangeTblEntry *fkrte;
const char *sep;
const char *fk_only;
+ const char *pk_only;
int i;
int save_nestlevel;
char workmembuf[32];
@@ -1769,7 +1779,7 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
/*----------
* The query string built is:
* SELECT fk.keycols FROM [ONLY] relname fk
- * LEFT OUTER JOIN ONLY pkrelname pk+ * LEFT OUTER JOIN [ONLY] pkrelname pk
* ON (pk.pkkeycol1=fk.keycol1 [AND ...])
* WHERE pk.pkkeycol1 IS NULL AND
* For MATCH SIMPLE:
@@ -1796,9 +1806,11 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
quoteRelationName(fkrelname, fk_rel);
fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?
"" : "ONLY ";
+ pk_only = pk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?+ "" : "ONLY ";
appendStringInfo(&querybuf,
- " FROM %s%s fk LEFT OUTER JOIN ONLY %s pk ON",- fk_only, fkrelname, pkrelname);+ " FROM %s%s fk LEFT OUTER JOIN %s%s pk ON",+ fk_only, fkrelname, pk_only, pkrelname);
strcpy(pkattname, "pk.");
strcpy(fkattname, "fk.");
@@ -1951,7 +1963,7 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
ri_ReportViolation(&fake_riinfo,
pk_rel, fk_rel,
tuple, tupdesc,
- RI_PLAN_CHECK_LOOKUPPK);+ RI_PLAN_CHECK_LOOKUPPK, false);
}
if (SPI_finish() != SPI_OK_FINISH)
@@ -1965,6 +1977,179 @@ RI_Initial_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)
return true;
}
+/* ----------+ * RI_Final_Check -+ *+ */+void+RI_Final_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel)+{+ const RI_ConstraintInfo *riinfo;+ StringInfoData querybuf;+ char pkrelname[MAX_QUOTED_REL_NAME_LEN];+ char fkrelname[MAX_QUOTED_REL_NAME_LEN];+ char pkattname[MAX_QUOTED_NAME_LEN + 3];+ char fkattname[MAX_QUOTED_NAME_LEN + 3];+ const char *sep;+ const char *fk_only;+ int spi_result;+ SPIPlanPtr qplan;+ int i;++ riinfo = ri_FetchConstraintInfo(trigger, fk_rel, false);++ /* XXX handle the non-permission case?? */++ /*----------+ * The query string built is:+ * SELECT fk.keycols FROM [ONLY] relname fk+ * JOIN pkrelname pk+ * ON (pk.pkkeycol1=fk.keycol1 [AND ...])+ * WHERE (<partition constraint>) AND+ * For MATCH SIMPLE:+ * (fk.keycol1 IS NOT NULL [AND ...])+ * For MATCH FULL:+ * (fk.keycol1 IS NOT NULL [OR ...])+ *+ * We attach COLLATE clauses to the operators when comparing columns+ * that have different collations.+ *----------+ */+ initStringInfo(&querybuf);+ appendStringInfoString(&querybuf, "SELECT ");+ sep = "";+ for (i = 0; i < riinfo->nkeys; i++)+ {+ quoteOneName(fkattname,+ RIAttName(fk_rel, riinfo->fk_attnums[i]));+ appendStringInfo(&querybuf, "%sfk.%s", sep, fkattname);+ sep = ", ";+ }++ quoteRelationName(pkrelname, pk_rel);+ quoteRelationName(fkrelname, fk_rel);+ fk_only = fk_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE ?+ "" : "ONLY ";+ appendStringInfo(&querybuf,+ " FROM %s%s fk JOIN %s pk ON",+ fk_only, fkrelname, pkrelname);+ strcpy(pkattname, "pk.");+ strcpy(fkattname, "fk.");+ sep = "(";+ for (i = 0; i < riinfo->nkeys; i++)+ {+ Oid pk_type = RIAttType(pk_rel, riinfo->pk_attnums[i]);+ Oid fk_type = RIAttType(fk_rel, riinfo->fk_attnums[i]);+ Oid pk_coll = RIAttCollation(pk_rel, riinfo->pk_attnums[i]);+ Oid fk_coll = RIAttCollation(fk_rel, riinfo->fk_attnums[i]);++ quoteOneName(pkattname + 3,+ RIAttName(pk_rel, riinfo->pk_attnums[i]));+ quoteOneName(fkattname + 3,+ RIAttName(fk_rel, riinfo->fk_attnums[i]));+ ri_GenerateQual(&querybuf, sep,+ pkattname, pk_type,+ riinfo->pf_eq_oprs[i],+ fkattname, fk_type);+ if (pk_coll != fk_coll)+ ri_GenerateQualCollation(&querybuf, pk_coll);+ sep = "AND";+ }++ appendStringInfo(&querybuf, ") WHERE %s AND (",+ pg_get_partconstrdef_string(RelationGetRelid(pk_rel),+ "pk"));++ sep = "";+ for (i = 0; i < riinfo->nkeys; i++)+ {+ quoteOneName(fkattname, RIAttName(fk_rel, riinfo->fk_attnums[i]));+ appendStringInfo(&querybuf,+ "%sfk.%s IS NOT NULL",+ sep, fkattname);+ switch (riinfo->confmatchtype)+ {+ case FKCONSTR_MATCH_SIMPLE:+ sep = " AND ";+ break;+ case FKCONSTR_MATCH_FULL:+ sep = " OR ";+ break;+ case FKCONSTR_MATCH_PARTIAL:+ ereport(ERROR,+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),+ errmsg("MATCH PARTIAL not yet implemented")));+ break;+ default:+ elog(ERROR, "unrecognized confmatchtype: %d",+ riinfo->confmatchtype);+ break;+ }+ }+ appendStringInfoChar(&querybuf, ')');++ /*+ * RI_Initial_Check changes work_mem here.+ */++ if (SPI_connect() != SPI_OK_CONNECT)+ elog(ERROR, "SPI_connect failed");++ /*+ * Generate the plan. We don't need to cache it, and there are no+ * arguments to the plan.+ */+ qplan = SPI_prepare(querybuf.data, 0, NULL);++ if (qplan == NULL)+ elog(ERROR, "SPI_prepare returned %s for %s",+ SPI_result_code_string(SPI_result), querybuf.data);++ /*+ * Run the plan. For safety we force a current snapshot to be used. (In+ * transaction-snapshot mode, this arguably violates transaction isolation+ * rules, but we really haven't got much choice.) We don't need to+ * register the snapshot, because SPI_execute_snapshot will see to it. We+ * need at most one tuple returned, so pass limit = 1.+ */+ spi_result = SPI_execute_snapshot(qplan,+ NULL, NULL,+ GetLatestSnapshot(),+ InvalidSnapshot,+ true, false, 1);++ /* Check result */+ if (spi_result != SPI_OK_SELECT)+ elog(ERROR, "SPI_execute_snapshot returned %s", SPI_result_code_string(spi_result));++ /* Did we find a tuple that would violate the constraint? */+ if (SPI_processed > 0)+ {+ HeapTuple tuple = SPI_tuptable->vals[0];+ TupleDesc tupdesc = SPI_tuptable->tupdesc;+ RI_ConstraintInfo fake_riinfo;++ /*+ * The columns to look at in the result tuple are 1..N, not whatever+ * they are in the fk_rel. Hack up riinfo so that ri_ReportViolation+ * will behave properly.+ *+ * In addition to this, we have to pass the correct tupdesc to+ * ri_ReportViolation, overriding its normal habit of using the pk_rel+ * or fk_rel's tupdesc.+ */+ memcpy(&fake_riinfo, riinfo, sizeof(RI_ConstraintInfo));+ for (i = 0; i < fake_riinfo.nkeys; i++)+ fake_riinfo.pk_attnums[i] = i + 1;++ ri_ReportViolation(&fake_riinfo, pk_rel, fk_rel,+ tuple, tupdesc, 0, true);+ }++ if (SPI_finish() != SPI_OK_FINISH)+ elog(ERROR, "SPI_finish failed");+}+
/* ----------
* Local functions below
@@ -2166,6 +2351,7 @@ ri_FetchConstraintInfo(Trigger *trigger, Relation trig_rel, bool rel_is_pk)
/* Find or create a hashtable entry for the constraint */
riinfo = ri_LoadConstraintInfo(constraintOid);
+#if 0
/* Do some easy cross-checks against the trigger call data */
if (rel_is_pk)
{
@@ -2174,6 +2360,7 @@ ri_FetchConstraintInfo(Trigger *trigger, Relation trig_rel, bool rel_is_pk)
elog(ERROR, "wrong pg_constraint entry for trigger \"%s\" on table \"%s\"",
trigger->tgname, RelationGetRelationName(trig_rel));
}
+#endif
return riinfo;
}
@@ -2479,7 +2666,7 @@ ri_PerformCheck(const RI_ConstraintInfo *riinfo,
pk_rel, fk_rel,
new_tuple ? new_tuple : old_tuple,
NULL,
- qkey->constr_queryno);+ qkey->constr_queryno, false);
return SPI_processed != 0;
}
@@ -2523,7 +2710,7 @@ static void
ri_ReportViolation(const RI_ConstraintInfo *riinfo,
Relation pk_rel, Relation fk_rel,
HeapTuple violator, TupleDesc tupdesc,
- int queryno)+ int queryno, bool partgone)
{
StringInfoData key_names;
StringInfoData key_values;
@@ -2563,9 +2750,13 @@ ri_ReportViolation(const RI_ConstraintInfo *riinfo,
*
* Check table-level permissions next and, failing that, column-level
* privileges.
+ *+ * When a partition at the referenced side is being detached/dropped, we+ * needn't check, since the user must be the table owner anyway.
*/
-- if (check_enable_rls(rel_oid, InvalidOid, true) != RLS_ENABLED)+ if (partgone)+ has_perm = true;+ else if (check_enable_rls(rel_oid, InvalidOid, true) != RLS_ENABLED)
{
aclresult = pg_class_aclcheck(rel_oid, GetUserId(), ACL_SELECT);
if (aclresult != ACLCHECK_OK)
@@ -2615,7 +2806,16 @@ ri_ReportViolation(const RI_ConstraintInfo *riinfo,
}
}
- if (onfk)+ if (partgone)+ ereport(ERROR,+ (errcode(ERRCODE_FOREIGN_KEY_VIOLATION),+ errmsg("removing partition \"%s\" violates foreign key constraint \"%s\"",+ RelationGetRelationName(pk_rel),+ NameStr(riinfo->conname)),+ errdetail("Key (%s)=(%s) still referenced from table \"%s\".",+ key_names.data, key_values.data,+ RelationGetRelationName(fk_rel))));+ else if (onfk)
ereport(ERROR,
(errcode(ERRCODE_FOREIGN_KEY_VIOLATION),
errmsg("insert or update on table \"%s\" violates foreign key constraint \"%s\"",
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.cindex 4857caecaa..b2deee6696 100644--- a/src/backend/utils/adt/ruleutils.c+++ b/src/backend/utils/adt/ruleutils.c@@ -1818,6 +1818,24 @@ pg_get_partition_constraintdef(PG_FUNCTION_ARGS)
}
/*
+ * pg_get_partconstrdef_string+ *+ * Returns the partition constraint as a C-string for the input relation, with+ * the given alias. No pretty-printing.+ */+char *+pg_get_partconstrdef_string(Oid partitionId, char *aliasname)+{+ Expr *constr_expr;+ List *context;++ constr_expr = get_partition_qual_relid(partitionId);+ context = deparse_context_for(aliasname, partitionId);++ return deparse_expression((Node *) constr_expr, context, true, false);+}++/*
* pg_get_constraintdef
*
* Returns the definition for the constraint, ie, everything that needs to
diff --git a/src/include/catalog/heap.h b/src/include/catalog/heap.hindex 56a341a622..f7cf97554a 100644--- a/src/include/catalog/heap.h+++ b/src/include/catalog/heap.h@@ -77,6 +77,8 @@ extern void heap_create_init_fork(Relation rel);
extern void heap_drop_with_catalog(Oid relid);
+extern void pre_drop_class_check(Oid relationId, Oid objectSubId);+
extern void heap_truncate(List *relids);
extern void heap_truncate_one_rel(Relation rel);
diff --git a/src/include/commands/tablecmds.h b/src/include/commands/tablecmds.hindex 2afcd5be44..350674bf67 100644--- a/src/include/commands/tablecmds.h+++ b/src/include/commands/tablecmds.h@@ -52,6 +52,8 @@ extern void AlterRelationNamespaceInternal(Relation classRel, Oid relOid,
extern void CheckTableNotInUse(Relation rel, const char *stmt);
+extern void CheckNoForeignKeyRefs(Relation partition, bool isDrop);+
extern void ExecuteTruncate(TruncateStmt *stmt);
extern void ExecuteTruncateGuts(List *explicit_rels, List *relids, List *relids_logged,
DropBehavior behavior, bool restart_seqs);
@@ -76,10 +78,6 @@ extern void find_composite_type_dependencies(Oid typeOid,
extern void check_of_type(HeapTuple typetuple);
-extern void createForeignKeyTriggers(Relation rel, Oid refRelOid,- Constraint *fkconstraint, Oid constraintOid,- Oid indexOid, bool create_action);-
extern void register_on_commit_action(Oid relid, OnCommitAction action);
extern void remove_on_commit_action(Oid relid);
diff --git a/src/include/commands/trigger.h b/src/include/commands/trigger.hindex 1031448c14..3183d8b14a 100644--- a/src/include/commands/trigger.h+++ b/src/include/commands/trigger.h@@ -263,6 +263,7 @@ extern bool RI_FKey_fk_upd_check_required(Trigger *trigger, Relation fk_rel,
HeapTuple old_row, HeapTuple new_row);
extern bool RI_Initial_Check(Trigger *trigger,
Relation fk_rel, Relation pk_rel);
+extern void RI_Final_Check(Trigger *trigger, Relation fk_rel, Relation pk_rel);
/* result values for RI_FKey_trigger_type: */
#define RI_TRIGGER_PK 1 /* is a trigger on the PK relation */
diff --git a/src/include/utils/ruleutils.h b/src/include/utils/ruleutils.hindex 9f9b029ab8..dd8ea7861d 100644--- a/src/include/utils/ruleutils.h+++ b/src/include/utils/ruleutils.h@@ -22,6 +22,7 @@ extern char *pg_get_indexdef_string(Oid indexrelid);
extern char *pg_get_indexdef_columns(Oid indexrelid, bool pretty);
extern char *pg_get_partkeydef_columns(Oid relid, bool pretty);
+extern char *pg_get_partconstrdef_string(Oid partitionId, char *aliasname);
extern char *pg_get_constraintdef_command(Oid constraintId);
extern char *deparse_expression(Node *expr, List *dpcontext,
diff --git a/src/test/regress/expected/foreign_key.out b/src/test/regress/expected/foreign_key.outindex 085c9aba11..66c083a8e1 100644--- a/src/test/regress/expected/foreign_key.out+++ b/src/test/regress/expected/foreign_key.out@@ -1431,19 +1431,6 @@ drop table pktable2, fktable2;
--
-- Foreign keys and partitioned tables
--
--- partitioned table in the referenced side are not allowed-CREATE TABLE fk_partitioned_pk (a int, b int, primary key (a, b))- PARTITION BY RANGE (a, b);--- verify with create table first ...-CREATE TABLE fk_notpartitioned_fk (a int, b int,- FOREIGN KEY (a, b) REFERENCES fk_partitioned_pk);-ERROR: cannot reference partitioned table "fk_partitioned_pk"--- and then with alter table.-CREATE TABLE fk_notpartitioned_fk_2 (a int, b int);-ALTER TABLE fk_notpartitioned_fk_2 ADD FOREIGN KEY (a, b)- REFERENCES fk_partitioned_pk;-ERROR: cannot reference partitioned table "fk_partitioned_pk"-DROP TABLE fk_partitioned_pk, fk_notpartitioned_fk_2;
-- Creation of a partitioned hierarchy with irregular definitions
CREATE TABLE fk_notpartitioned_pk (fdrop1 int, a int, fdrop2 int, b int,
PRIMARY KEY (a, b));
@@ -1479,10 +1466,10 @@ DETAIL: This feature is not yet supported on partitioned tables.
-- these inserts, targeting both the partition directly as well as the
-- partitioned table, should all fail
INSERT INTO fk_partitioned_fk (a,b) VALUES (500, 501);
-ERROR: insert or update on table "fk_partitioned_fk_1" violates foreign key constraint "fk_partitioned_fk_a_fkey"+ERROR: insert or update on table "fk_partitioned_fk_1" violates foreign key constraint "fk_partitioned_fk_1_a_fkey"
DETAIL: Key (a, b)=(500, 501) is not present in table "fk_notpartitioned_pk".
INSERT INTO fk_partitioned_fk_1 (a,b) VALUES (500, 501);
-ERROR: insert or update on table "fk_partitioned_fk_1" violates foreign key constraint "fk_partitioned_fk_a_fkey"+ERROR: insert or update on table "fk_partitioned_fk_1" violates foreign key constraint "fk_partitioned_fk_1_a_fkey"
DETAIL: Key (a, b)=(500, 501) is not present in table "fk_notpartitioned_pk".
INSERT INTO fk_partitioned_fk (a,b) VALUES (1500, 1501);
ERROR: insert or update on table "fk_partitioned_fk_2" violates foreign key constraint "fk_partitioned_fk_a_fkey"
@@ -1609,7 +1596,7 @@ INSERT INTO fk_partitioned_fk_3 (a, b) VALUES (2502, 2503);
-- this fails, because the defaults for the referencing table are not present
-- in the referenced table:
UPDATE fk_notpartitioned_pk SET a = 1500 WHERE a = 2502;
-ERROR: insert or update on table "fk_partitioned_fk_3" violates foreign key constraint "fk_partitioned_fk_a_fkey"+ERROR: insert or update on table "fk_partitioned_fk_3" violates foreign key constraint "fk_partitioned_fk_3_a_fkey"
DETAIL: Key (a, b)=(2501, 142857) is not present in table "fk_notpartitioned_pk".
-- but inserting the row we can make it work:
INSERT INTO fk_notpartitioned_pk VALUES (2501, 142857);
@@ -1781,3 +1768,152 @@ INSERT INTO fk_notpartitioned_pk VALUES (1600, 601), (1600, 1601);
ALTER TABLE fk_partitioned_fk ATTACH PARTITION fk_partitioned_fk_2
FOR VALUES IN (1600);
-- leave these tables around intentionally
+-- Test a partitioned table as referenced table.+-- Verify basic functionality with a regular partition creation and a partition+-- with a different column layout, as well as partitions+-- added (created and attached) after creating the foreign key.+create schema regress_fk;+set search_path to regress_fk;+create table pk (a int primary key) partition by range (a);+create table pk1 partition of pk for values from (0) to (1000);+create table pk2 (b int, a int);+alter table pk2 drop column b;+alter table pk2 alter a set not null;+alter table pk attach partition pk2 for values from (1000) to (2000);+create table fk (a int) partition by range (a);+create table fk1 partition of fk for values from (0) to (750);+alter table fk add foreign key (a) references pk;+create table fk2 (b int, a int) ;+alter table fk2 drop column b;+alter table fk attach partition fk2 for values from (750) to (3500);+create table pk3 partition of pk for values from (2000) to (3000);+create table pk4 (like pk);+alter table pk attach partition pk4 for values from (3000) to (4000);+create table pk5 (like pk) partition by range (a);+create table pk51 partition of pk5 for values from (4000) to (4500);+create table pk52 partition of pk5 for values from (4500) to (5000);+alter table pk attach partition pk5 for values from (4000) to (5000);+create table fk3 partition of fk for values from (3500) to (5000);+-- these should fail: referenced value not present+insert into fk values (1);+ERROR: insert or update on table "fk1" violates foreign key constraint "fk1_a_fkey"+DETAIL: Key (a)=(1) is not present in table "pk".+insert into fk values (1000);+ERROR: insert or update on table "fk2" violates foreign key constraint "fk_a_fkey"+DETAIL: Key (a)=(1000) is not present in table "pk".+insert into fk values (2000);+ERROR: insert or update on table "fk2" violates foreign key constraint "fk_a_fkey"+DETAIL: Key (a)=(2000) is not present in table "pk".+insert into fk values (3000);+ERROR: insert or update on table "fk2" violates foreign key constraint "fk_a_fkey"+DETAIL: Key (a)=(3000) is not present in table "pk".+insert into fk values (4000);+ERROR: insert or update on table "fk3" violates foreign key constraint "fk_a_fkey"+DETAIL: Key (a)=(4000) is not present in table "pk".+insert into fk values (4500);+ERROR: insert or update on table "fk3" violates foreign key constraint "fk_a_fkey"+DETAIL: Key (a)=(4500) is not present in table "pk".+-- insert into the referenced table, now they should work+insert into pk values (1), (1000), (2000), (3000), (4000), (4500);+insert into fk values (1), (1000), (2000), (3000), (4000), (4500);+-- should fail: referencing value present+delete from pk where a = 1;+ERROR: update or delete on table "pk1" violates foreign key constraint "fk_a_fkey1" on table "fk"+DETAIL: Key (a)=(1) is still referenced from table "fk".+delete from pk where a = 1000;+ERROR: update or delete on table "pk2" violates foreign key constraint "fk_a_fkey2" on table "fk"+DETAIL: Key (a)=(1000) is still referenced from table "fk".+delete from pk where a = 2000;+ERROR: update or delete on table "pk3" violates foreign key constraint "fk_a_fkey3" on table "fk"+DETAIL: Key (a)=(2000) is still referenced from table "fk".+delete from pk where a = 3000;+ERROR: update or delete on table "pk4" violates foreign key constraint "fk_a_fkey4" on table "fk"+DETAIL: Key (a)=(3000) is still referenced from table "fk".+delete from pk where a = 4000;+ERROR: update or delete on table "pk51" violates foreign key constraint "fk_a_fkey6" on table "fk"+DETAIL: Key (a)=(4000) is still referenced from table "fk".+delete from pk where a = 4500;+ERROR: update or delete on table "pk52" violates foreign key constraint "fk_a_fkey7" on table "fk"+DETAIL: Key (a)=(4500) is still referenced from table "fk".+update pk set a = 2 where a = 1;+ERROR: update or delete on table "pk1" violates foreign key constraint "fk_a_fkey1" on table "fk"+DETAIL: Key (a)=(1) is still referenced from table "fk".+update pk set a = 1002 where a = 1000;+ERROR: update or delete on table "pk2" violates foreign key constraint "fk_a_fkey2" on table "fk"+DETAIL: Key (a)=(1000) is still referenced from table "fk".+update pk set a = 2002 where a = 2000;+ERROR: update or delete on table "pk3" violates foreign key constraint "fk_a_fkey3" on table "fk"+DETAIL: Key (a)=(2000) is still referenced from table "fk".+update pk set a = 3002 where a = 3000;+ERROR: update or delete on table "pk4" violates foreign key constraint "fk_a_fkey4" on table "fk"+DETAIL: Key (a)=(3000) is still referenced from table "fk".+update pk set a = 4002 where a = 4000;+ERROR: update or delete on table "pk51" violates foreign key constraint "fk_a_fkey6" on table "fk"+DETAIL: Key (a)=(4000) is still referenced from table "fk".+update pk set a = 4502 where a = 4500;+ERROR: update or delete on table "pk52" violates foreign key constraint "fk_a_fkey7" on table "fk"+DETAIL: Key (a)=(4500) is still referenced from table "fk".+-- now they should work+delete from fk;+update pk set a = 2 where a = 1;+delete from pk where a = 2;+update pk set a = 1002 where a = 1000;+delete from pk where a = 1002;+update pk set a = 2002 where a = 2000;+delete from pk where a = 2002;+update pk set a = 3002 where a = 3000;+delete from pk where a = 3002;+update pk set a = 4002 where a = 4000;+delete from pk where a = 4002;+update pk set a = 4502 where a = 4500;+delete from pk where a = 4502;+-- dropping/detaching partitions is prevented if that would break+-- a foreign key's existing data+create table droppk (a int primary key) partition by range (a);+create table droppk1 partition of droppk for values from (0) to (1000);+create table droppk_d partition of droppk default;+create table droppk2 partition of droppk for values from (1000) to (2000)+ partition by range (a);+create table droppk21 partition of droppk2 for values from (1000) to (1400);+create table droppk2_d partition of droppk2 default;+insert into droppk values (1), (1000), (1500), (2000);+create table dropfk (a int references droppk);+insert into dropfk values (1), (1000), (1500), (2000);+-- these should all fail+alter table droppk detach partition droppk_d;+ERROR: removing partition "droppk_d" violates foreign key constraint "dropfk_a_fkey5"+DETAIL: Key (a)=(2000) still referenced from table "dropfk".+alter table droppk2 detach partition droppk2_d;+ERROR: removing partition "droppk2_d" violates foreign key constraint "dropfk_a_fkey4"+DETAIL: Key (a)=(1500) still referenced from table "dropfk".+alter table droppk detach partition droppk1;+ERROR: removing partition "droppk1" violates foreign key constraint "dropfk_a_fkey1"+DETAIL: Key (a)=(1) still referenced from table "dropfk".+alter table droppk detach partition droppk2;+ERROR: removing partition "droppk2" violates foreign key constraint "dropfk_a_fkey2"+DETAIL: Key (a)=(1000) still referenced from table "dropfk".+alter table droppk2 detach partition droppk21;+ERROR: removing partition "droppk21" violates foreign key constraint "dropfk_a_fkey3"+DETAIL: Key (a)=(1000) still referenced from table "dropfk".+drop table droppk_d;+ERROR: removing partition "droppk_d" violates foreign key constraint "dropfk_a_fkey5"+DETAIL: Key (a)=(2000) still referenced from table "dropfk".+drop table droppk2_d;+ERROR: removing partition "droppk2_d" violates foreign key constraint "dropfk_a_fkey4"+DETAIL: Key (a)=(1500) still referenced from table "dropfk".+drop table droppk1;+ERROR: removing partition "droppk1" violates foreign key constraint "dropfk_a_fkey1"+DETAIL: Key (a)=(1) still referenced from table "dropfk".+drop table droppk2;+ERROR: removing partition "droppk2_d" violates foreign key constraint "dropfk_a_fkey4"+DETAIL: Key (a)=(1500) still referenced from table "dropfk".+drop table droppk21;+ERROR: removing partition "droppk21" violates foreign key constraint "dropfk_a_fkey3"+DETAIL: Key (a)=(1000) still referenced from table "dropfk".+delete from dropfk;+-- now they should all work+drop table droppk_d;+drop table droppk2_d;+drop table droppk1;+alter table droppk2 detach partition droppk21;+drop table droppk2;diff --git a/src/test/regress/sql/foreign_key.sql b/src/test/regress/sql/foreign_key.sqlindex 068ab2aab7..6416523b94 100644--- a/src/test/regress/sql/foreign_key.sql+++ b/src/test/regress/sql/foreign_key.sql@@ -1071,18 +1071,6 @@ drop table pktable2, fktable2;
-- Foreign keys and partitioned tables
--
--- partitioned table in the referenced side are not allowed-CREATE TABLE fk_partitioned_pk (a int, b int, primary key (a, b))- PARTITION BY RANGE (a, b);--- verify with create table first ...-CREATE TABLE fk_notpartitioned_fk (a int, b int,- FOREIGN KEY (a, b) REFERENCES fk_partitioned_pk);--- and then with alter table.-CREATE TABLE fk_notpartitioned_fk_2 (a int, b int);-ALTER TABLE fk_notpartitioned_fk_2 ADD FOREIGN KEY (a, b)- REFERENCES fk_partitioned_pk;-DROP TABLE fk_partitioned_pk, fk_notpartitioned_fk_2;-
-- Creation of a partitioned hierarchy with irregular definitions
CREATE TABLE fk_notpartitioned_pk (fdrop1 int, a int, fdrop2 int, b int,
PRIMARY KEY (a, b));
@@ -1289,3 +1277,105 @@ ALTER TABLE fk_partitioned_fk ATTACH PARTITION fk_partitioned_fk_2
FOR VALUES IN (1600);
-- leave these tables around intentionally
++-- Test a partitioned table as referenced table.+-- Verify basic functionality with a regular partition creation and a partition+-- with a different column layout, as well as partitions+-- added (created and attached) after creating the foreign key.+create schema regress_fk;+set search_path to regress_fk;++create table pk (a int primary key) partition by range (a);+create table pk1 partition of pk for values from (0) to (1000);+create table pk2 (b int, a int);+alter table pk2 drop column b;+alter table pk2 alter a set not null;+alter table pk attach partition pk2 for values from (1000) to (2000);++create table fk (a int) partition by range (a);+create table fk1 partition of fk for values from (0) to (750);+alter table fk add foreign key (a) references pk;+create table fk2 (b int, a int) ;+alter table fk2 drop column b;+alter table fk attach partition fk2 for values from (750) to (3500);++create table pk3 partition of pk for values from (2000) to (3000);+create table pk4 (like pk);+alter table pk attach partition pk4 for values from (3000) to (4000);++create table pk5 (like pk) partition by range (a);+create table pk51 partition of pk5 for values from (4000) to (4500);+create table pk52 partition of pk5 for values from (4500) to (5000);+alter table pk attach partition pk5 for values from (4000) to (5000);++create table fk3 partition of fk for values from (3500) to (5000);++-- these should fail: referenced value not present+insert into fk values (1);+insert into fk values (1000);+insert into fk values (2000);+insert into fk values (3000);+insert into fk values (4000);+insert into fk values (4500);+-- insert into the referenced table, now they should work+insert into pk values (1), (1000), (2000), (3000), (4000), (4500);+insert into fk values (1), (1000), (2000), (3000), (4000), (4500);++-- should fail: referencing value present+delete from pk where a = 1;+delete from pk where a = 1000;+delete from pk where a = 2000;+delete from pk where a = 3000;+delete from pk where a = 4000;+delete from pk where a = 4500;+update pk set a = 2 where a = 1;+update pk set a = 1002 where a = 1000;+update pk set a = 2002 where a = 2000;+update pk set a = 3002 where a = 3000;+update pk set a = 4002 where a = 4000;+update pk set a = 4502 where a = 4500;+-- now they should work+delete from fk;+update pk set a = 2 where a = 1;+delete from pk where a = 2;+update pk set a = 1002 where a = 1000;+delete from pk where a = 1002;+update pk set a = 2002 where a = 2000;+delete from pk where a = 2002;+update pk set a = 3002 where a = 3000;+delete from pk where a = 3002;+update pk set a = 4002 where a = 4000;+delete from pk where a = 4002;+update pk set a = 4502 where a = 4500;+delete from pk where a = 4502;++-- dropping/detaching partitions is prevented if that would break+-- a foreign key's existing data+create table droppk (a int primary key) partition by range (a);+create table droppk1 partition of droppk for values from (0) to (1000);+create table droppk_d partition of droppk default;+create table droppk2 partition of droppk for values from (1000) to (2000)+ partition by range (a);+create table droppk21 partition of droppk2 for values from (1000) to (1400);+create table droppk2_d partition of droppk2 default;+insert into droppk values (1), (1000), (1500), (2000);+create table dropfk (a int references droppk);+insert into dropfk values (1), (1000), (1500), (2000);+-- these should all fail+alter table droppk detach partition droppk_d;+alter table droppk2 detach partition droppk2_d;+alter table droppk detach partition droppk1;+alter table droppk detach partition droppk2;+alter table droppk2 detach partition droppk21;+drop table droppk_d;+drop table droppk2_d;+drop table droppk1;+drop table droppk2;+drop table droppk21;+delete from dropfk;+-- now they should all work+drop table droppk_d;+drop table droppk2_d;+drop table droppk1;+alter table droppk2 detach partition droppk21;+drop table droppk2;--
2.11.0
[text/x-diff] 0005-Have-psql-not-display-redundant-FKs.patch (9.5K, ../20181130191216.5xcxcsx3ascgqayv@alvherre.pgsql/6-0005-Have-psql-not-display-redundant-FKs.patch)
download | inline diff:
From 87053b456eae36811d643e5ce8c1b03eae5b7091 Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Thu, 22 Nov 2018 19:30:22 -0300
Subject: [PATCH 5/5] Have psql not display redundant FKs
---
src/bin/psql/describe.c | 61 +++++++++++++++++++++++--------
src/test/regress/expected/foreign_key.out | 16 ++++----
2 files changed, 53 insertions(+), 24 deletions(-)
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.cindex 0a181b01d9..1ed73afaa8 100644--- a/src/bin/psql/describe.c+++ b/src/bin/psql/describe.c@@ -2382,17 +2382,28 @@ describeOneTableDetails(const char *schemaname,
/*
* Print foreign-key constraints (there are none if no triggers,
* except if the table is partitioned, in which case the triggers
- * appear in the partitions)+ * appear in the partitions). Note we put the constraints defined+ * in this table first, followed by the constraints defined in+ * ancestor partitioned tables. XXX maybe should display in hierarchy+ * order.
*/
if (tableinfo.hastriggers ||
tableinfo.relkind == RELKIND_PARTITIONED_TABLE)
{
printfPQExpBuffer(&buf,
- "SELECT conname,\n"- " pg_catalog.pg_get_constraintdef(r.oid, true) as condef\n"- "FROM pg_catalog.pg_constraint r\n"- "WHERE r.conrelid = '%s' AND r.contype = 'f' ORDER BY 1;",- oid);+ "WITH RECURSIVE constraints (oid, parent, conname, conrelid) as (\n"+ " SELECT oid, conparentid, conname, conrelid\n"+ " FROM pg_catalog.pg_constraint WHERE contype = 'f' AND conrelid = '%s'\n"+ " UNION\n"+ " SELECT pc.oid, pc.conparentid, pc.conname, pc.conrelid\n"+ " FROM constraints, pg_constraint pc\n"+ " WHERE pc.oid = constraints.parent\n"+ ") SELECT conrelid = '%s' as conislocal, conname,\n"+ " pg_catalog.pg_get_constraintdef(oid), conrelid::pg_catalog.regclass\n"+ " FROM constraints\n"+ " WHERE parent = 0\n"+ "ORDER BY conislocal DESC, conname;",+ oid, oid);
result = PSQLexec(buf.data);
if (!result)
goto error_return;
@@ -2404,10 +2415,20 @@ describeOneTableDetails(const char *schemaname,
printTableAddFooter(&cont, _("Foreign-key constraints:"));
for (i = 0; i < tuples; i++)
{
- /* untranslated constraint name and def */- printfPQExpBuffer(&buf, " \"%s\" %s",- PQgetvalue(result, i, 0),- PQgetvalue(result, i, 1));+ /*+ * Print untranslated constraint name and definition.+ * Use a "TABLE tab" prefix when the constraint is+ * defined in a parent partitioned table.+ */+ if (strcmp(PQgetvalue(result, i, 0), "f") == 0)+ printfPQExpBuffer(&buf, " TABLE \"%s\" CONSTRAINT \"%s\" %s",+ PQgetvalue(result, i, 3),+ PQgetvalue(result, i, 1),+ PQgetvalue(result, i, 2));+ else+ printfPQExpBuffer(&buf, " \"%s\" %s",+ PQgetvalue(result, i, 1),+ PQgetvalue(result, i, 2));
printTableAddFooter(&cont, buf.data);
}
@@ -2416,13 +2437,21 @@ describeOneTableDetails(const char *schemaname,
}
/* print incoming foreign-key references (none if no triggers) */
- if (tableinfo.hastriggers)+ if (tableinfo.hastriggers ||+ tableinfo.relkind == RELKIND_PARTITIONED_TABLE)
{
printfPQExpBuffer(&buf,
- "SELECT conname, conrelid::pg_catalog.regclass,\n"- " pg_catalog.pg_get_constraintdef(c.oid, true) as condef\n"- "FROM pg_catalog.pg_constraint c\n"- "WHERE c.confrelid = '%s' AND c.contype = 'f' ORDER BY 1;",+ "WITH RECURSIVE constraints (oid, parent, conname, conrelid, confrelid) as (\n"+ " SELECT oid, conparentid, conname, conrelid, confrelid\n"+ " FROM pg_catalog.pg_constraint WHERE contype = 'f' AND confrelid = '%s'\n"+ " UNION\n"+ " SELECT pc.oid, pc.conparentid, pc.conname, pc.conrelid, pc.confrelid\n"+ " FROM constraints, pg_constraint pc\n"+ " WHERE pc.oid = constraints.parent\n"+ ") SELECT conrelid::pg_catalog.regclass, conname, pg_catalog.pg_get_constraintdef(oid)\n"+ " FROM constraints\n"+ " WHERE parent = 0\n"+ "ORDER BY conname;",
oid);
result = PSQLexec(buf.data);
if (!result)
@@ -2436,8 +2465,8 @@ describeOneTableDetails(const char *schemaname,
for (i = 0; i < tuples; i++)
{
printfPQExpBuffer(&buf, " TABLE \"%s\" CONSTRAINT \"%s\" %s",
- PQgetvalue(result, i, 1),
PQgetvalue(result, i, 0),
+ PQgetvalue(result, i, 1),
PQgetvalue(result, i, 2));
printTableAddFooter(&cont, buf.data);
diff --git a/src/test/regress/expected/foreign_key.out b/src/test/regress/expected/foreign_key.outindex 66c083a8e1..5de2449119 100644--- a/src/test/regress/expected/foreign_key.out+++ b/src/test/regress/expected/foreign_key.out@@ -1667,7 +1667,7 @@ ALTER TABLE fk_partitioned_fk ATTACH PARTITION fk_partitioned_fk_2 FOR VALUES IN
a | 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+ TABLE "fk_partitioned_fk" CONSTRAINT "fk_partitioned_fk_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE
DROP TABLE fk_partitioned_fk_2;
CREATE TABLE fk_partitioned_fk_4 (a int, b int, FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE) PARTITION BY RANGE (b, a);
@@ -1687,7 +1687,7 @@ ALTER TABLE fk_partitioned_fk ATTACH PARTITION fk_partitioned_fk_4 FOR VALUES IN
Partition of: fk_partitioned_fk FOR VALUES IN (3500, 3502)
Partition key: RANGE (b, a)
Foreign-key constraints:
- "fk_partitioned_fk_4_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE+ TABLE "fk_partitioned_fk" CONSTRAINT "fk_partitioned_fk_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE
Number of partitions: 2 (Use \d+ to list them.)
\d fk_partitioned_fk_4_1
@@ -1698,7 +1698,7 @@ Number of partitions: 2 (Use \d+ to list them.)
b | integer | | |
Partition of: fk_partitioned_fk_4 FOR VALUES FROM (1, 1) TO (100, 100)
Foreign-key constraints:
- "fk_partitioned_fk_4_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE+ TABLE "fk_partitioned_fk" CONSTRAINT "fk_partitioned_fk_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE
-- this one has an FK with mismatched properties
\d fk_partitioned_fk_4_2
@@ -1710,7 +1710,7 @@ Foreign-key constraints:
Partition of: fk_partitioned_fk_4 FOR VALUES FROM (100, 100) TO (1000, 1000)
Foreign-key constraints:
"fk_partitioned_fk_4_2_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE SET NULL
- "fk_partitioned_fk_4_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE+ TABLE "fk_partitioned_fk" CONSTRAINT "fk_partitioned_fk_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE
CREATE TABLE fk_partitioned_fk_5 (a int, b int,
FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE DEFERRABLE,
@@ -1734,7 +1734,7 @@ Partition key: RANGE (a)
Foreign-key constraints:
"fk_partitioned_fk_5_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE DEFERRABLE
"fk_partitioned_fk_5_a_fkey1" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) MATCH FULL ON UPDATE CASCADE ON DELETE CASCADE
- "fk_partitioned_fk_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE+ TABLE "fk_partitioned_fk" CONSTRAINT "fk_partitioned_fk_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE
Number of partitions: 1 (Use \d+ to list them.)
-- verify that it works to reattaching a child with multiple candidate
@@ -1750,9 +1750,9 @@ ALTER TABLE fk_partitioned_fk_5 ATTACH PARTITION fk_partitioned_fk_5_1 FOR VALUE
Partition of: fk_partitioned_fk_5 FOR VALUES FROM (0) TO (10)
Foreign-key constraints:
"fk_partitioned_fk_5_1_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b)
- "fk_partitioned_fk_5_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE DEFERRABLE- "fk_partitioned_fk_5_a_fkey1" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) MATCH FULL ON UPDATE CASCADE ON DELETE CASCADE- "fk_partitioned_fk_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE+ TABLE "fk_partitioned_fk_5" CONSTRAINT "fk_partitioned_fk_5_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE DEFERRABLE+ TABLE "fk_partitioned_fk_5" CONSTRAINT "fk_partitioned_fk_5_a_fkey1" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) MATCH FULL ON UPDATE CASCADE ON DELETE CASCADE+ TABLE "fk_partitioned_fk" CONSTRAINT "fk_partitioned_fk_a_fkey" FOREIGN KEY (a, b) REFERENCES fk_notpartitioned_pk(a, b) ON UPDATE CASCADE ON DELETE CASCADE
-- verify that attaching a table checks that the existing data satisfies the
-- constraint
--
2.11.0
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: alvherre@2ndquadrant.com
Subject: Re: partitioned tables referenced by FKs
In-Reply-To: <20181130191216.5xcxcsx3ascgqayv@alvherre.pgsql>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox