Create action triggers when partitions are detached

Detaching a partition from a partitioned table that's constrained by
foreign keys requires additional action triggers on the referenced side;
otherwise, DELETE/UPDATE actions there fail to notice rows in the table
that was partition, and so are incorrectly allowed through.  With this
commit, those triggers are now created.  Conversely, when a table that
has a foreign key is attached as a partition to a table that also has
the same foreign key, those action triggers are no longer needed, so we
remove them.

Add a minimal test case verifying (part of) this.

Authors: Amit Langote, Álvaro Herrera
Discussion: https://postgr.es/m/f2b8ead5-4131-d5a8-8016-2ea0a31250af@lab.ntt.co.jp

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 9893c3521705bca037376274f47fd264764a6808..5a62adbafcd885913e7411f3f9e9310d4adae07a 100644 (file)
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -7945,6 +7945,10 @@ CloneFkReferencing(Relation pg_constraint, Relation parentRel,
                        ForeignKeyCacheInfo *fk = lfirst_node(ForeignKeyCacheInfo, cell);
                        Form_pg_constraint partConstr;
                        HeapTuple       partcontup;
+                       Relation        trigrel;
+                       HeapTuple       trigtup;
+                       SysScanDesc scan;
+                       ScanKeyData key;
 
                        attach_it = true;
 
@@ -7996,7 +8000,39 @@ CloneFkReferencing(Relation pg_constraint, Relation parentRel,
 
                        ReleaseSysCache(partcontup);
 
-                       /* looks good!  Attach this constraint */
+                       /*
+                        * Looks good!  Attach this constraint.  Note that the action
+                        * triggers are no longer needed, so remove them.  We identify
+                        * them because they have our constraint OID, as well as being
+                        * on the referenced rel.
+                        */
+                       trigrel = heap_open(TriggerRelationId, RowExclusiveLock);
+                       ScanKeyInit(&key,
+                                               Anum_pg_trigger_tgconstraint,
+                                               BTEqualStrategyNumber, F_OIDEQ,
+                                               ObjectIdGetDatum(fk->conoid));
+
+                       scan = systable_beginscan(trigrel, TriggerConstraintIndexId, true,
+                                                                         NULL, 1, &key);
+                       while ((trigtup = systable_getnext(scan)) != NULL)
+                       {
+                               Form_pg_trigger trgform = (Form_pg_trigger) GETSTRUCT(trigtup);
+
+                               if (trgform->tgconstrrelid != fk->conrelid)
+                                       continue;
+                               if (trgform->tgrelid != fk->confrelid)
+                                       continue;
+
+                               deleteDependencyRecordsForClass(TriggerRelationId,
+                                                                                               HeapTupleGetOid(trigtup),
+                                                                                               ConstraintRelationId,
+                                                                                               DEPENDENCY_INTERNAL);
+                               CatalogTupleDelete(trigrel, &trigtup->t_self);
+                       }
+
+                       systable_endscan(scan);
+                       heap_close(trigrel, RowExclusiveLock);
+
                        ConstraintSetParentConstraint(fk->conoid, parentConstrOid);
                        CommandCounterIncrement();
                        attach_it = true;
@@ -15211,19 +15247,50 @@ ATExecDetachPartition(Relation rel, RangeVar *name)
        }
        heap_close(classRel, RowExclusiveLock);
 
-       /* Detach foreign keys */
+       /*
+        * Detach any foreign keys that are inherited.  This includes creating
+        * additional action triggers.
+        */
        fks = copyObject(RelationGetFKeyList(partRel));
        foreach(cell, fks)
        {
                ForeignKeyCacheInfo *fk = lfirst(cell);
                HeapTuple       contup;
+               Form_pg_constraint conform;
+               Constraint *fkconstraint;
 
                contup = SearchSysCache1(CONSTROID, ObjectIdGetDatum(fk->conoid));
                if (!contup)
                        elog(ERROR, "cache lookup failed for constraint %u", fk->conoid);
+               conform = (Form_pg_constraint) GETSTRUCT(contup);
+
+               /* consider only the inherited foreign keys */
+               if (conform->contype != CONSTRAINT_FOREIGN ||
+                       !OidIsValid(conform->conparentid))
+               {
+                       ReleaseSysCache(contup);
+                       continue;
+               }
 
+               /* unset conparentid and adjust conislocal, coninhcount, etc. */
                ConstraintSetParentConstraint(fk->conoid, InvalidOid);
 
+               /*
+                * Make the action triggers on the referenced relation.  When this was
+                * a partition the action triggers pointed to the parent rel (they
+                * still do), but now we need separate ones of our own.
+                */
+               fkconstraint = makeNode(Constraint);
+               fkconstraint->conname = pstrdup(NameStr(conform->conname));
+               fkconstraint->fk_upd_action = conform->confupdtype;
+               fkconstraint->fk_del_action = conform->confdeltype;
+               fkconstraint->deferrable = conform->condeferrable;
+               fkconstraint->initdeferred = conform->condeferred;
+
+               createForeignKeyActionTriggers(partRel, conform->confrelid,
+                                                                          fkconstraint, fk->conoid,
+                                                                          conform->conindid);
+
                ReleaseSysCache(contup);
        }
        list_free_deep(fks);

diff --git a/src/test/regress/expected/foreign_key.out b/src/test/regress/expected/foreign_key.out
index 32a3611c9462049d75f244687ee0ac2c0f976c6d..4fd23413882cf976a091d71b7a9e394cfe948a6a 100644 (file)
--- a/src/test/regress/expected/foreign_key.out
+++ b/src/test/regress/expected/foreign_key.out
@@ -1860,7 +1860,31 @@ alter table fkpart0.fk_part_56 drop constraint fk_part_a_fkey;
 ERROR:  cannot drop inherited constraint "fk_part_a_fkey" of relation "fk_part_56"
 alter table fkpart0.fk_part_56_5 drop constraint fk_part_a_fkey;
 ERROR:  cannot drop inherited constraint "fk_part_a_fkey" of relation "fk_part_56_5"
+-- verify that attaching and detaching partitions maintains the right set of
+-- triggers
+create schema fkpart1
+  create table pkey (a int primary key)
+  create table fk_part (a int) partition by list (a)
+  create table fk_part_1 partition of fk_part for values in (1) partition by list (a)
+  create table fk_part_1_1 partition of fk_part_1 for values in (1);
+alter table fkpart1.fk_part add foreign key (a) references fkpart1.pkey;
+insert into fkpart1.fk_part values (1);                -- should fail
+ERROR:  insert or update on table "fk_part_1_1" violates foreign key constraint "fk_part_a_fkey"
+DETAIL:  Key (a)=(1) is not present in table "pkey".
+insert into fkpart1.pkey values (1);
+insert into fkpart1.fk_part values (1);
+delete from fkpart1.pkey where a = 1;          -- should fail
+ERROR:  update or delete on table "pkey" violates foreign key constraint "fk_part_a_fkey" on table "fk_part"
+DETAIL:  Key (a)=(1) is still referenced from table "fk_part".
+alter table fkpart1.fk_part detach partition fkpart1.fk_part_1;
+create table fkpart1.fk_part_1_2 partition of fkpart1.fk_part_1 for values in (2);
+insert into fkpart1.fk_part_1 values (2);      -- should fail
+ERROR:  insert or update on table "fk_part_1_2" violates foreign key constraint "fk_part_a_fkey"
+DETAIL:  Key (a)=(2) is not present in table "pkey".
+delete from fkpart1.pkey where a = 1;
+ERROR:  update or delete on table "pkey" violates foreign key constraint "fk_part_a_fkey" on table "fk_part_1"
+DETAIL:  Key (a)=(1) is still referenced from table "fk_part_1".
 \set VERBOSITY terse   \\ -- suppress cascade details
-drop schema fkpart0 cascade;
-NOTICE:  drop cascades to 2 other objects
+drop schema fkpart0, fkpart1 cascade;
+NOTICE:  drop cascades to 5 other objects
 \set VERBOSITY default

diff --git a/src/test/regress/sql/foreign_key.sql b/src/test/regress/sql/foreign_key.sql
index c120af739781a4ef564c157817d17f5f1bec56eb..dd0be01c7747c030fd1af67fa1424d24b48d5d1c 100644 (file)
--- a/src/test/regress/sql/foreign_key.sql
+++ b/src/test/regress/sql/foreign_key.sql
@@ -1324,6 +1324,23 @@ create table fkpart0.fk_part_56_5 partition of fkpart0.fk_part_56
 alter table fkpart0.fk_part_56 drop constraint fk_part_a_fkey;
 alter table fkpart0.fk_part_56_5 drop constraint fk_part_a_fkey;
 
+-- verify that attaching and detaching partitions maintains the right set of
+-- triggers
+create schema fkpart1
+  create table pkey (a int primary key)
+  create table fk_part (a int) partition by list (a)
+  create table fk_part_1 partition of fk_part for values in (1) partition by list (a)
+  create table fk_part_1_1 partition of fk_part_1 for values in (1);
+alter table fkpart1.fk_part add foreign key (a) references fkpart1.pkey;
+insert into fkpart1.fk_part values (1);                -- should fail
+insert into fkpart1.pkey values (1);
+insert into fkpart1.fk_part values (1);
+delete from fkpart1.pkey where a = 1;          -- should fail
+alter table fkpart1.fk_part detach partition fkpart1.fk_part_1;
+create table fkpart1.fk_part_1_2 partition of fkpart1.fk_part_1 for values in (2);
+insert into fkpart1.fk_part_1 values (2);      -- should fail
+delete from fkpart1.pkey where a = 1;
+
 \set VERBOSITY terse   \\ -- suppress cascade details
-drop schema fkpart0 cascade;
+drop schema fkpart0, fkpart1 cascade;
 \set VERBOSITY default