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 ba35c028943bdb86bf55478524961539c87c629f..fbd2d101c190599cfb55c479b585396b8a045ddf 100644 (file)
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -7851,6 +7851,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;
 
@@ -7902,7 +7906,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,
+                                               trgform->oid,
+                                               ConstraintRelationId,
+                                               DEPENDENCY_INTERNAL);
+               CatalogTupleDelete(trigrel, &trigtup->t_self);
+           }
+
+           systable_endscan(scan);
+           heap_close(trigrel, RowExclusiveLock);
+
            ConstraintSetParentConstraint(fk->conoid, parentConstrOid);
            CommandCounterIncrement();
            attach_it = true;
@@ -15073,19 +15109,50 @@ ATExecDetachPartition(Relation rel, RangeVar *name)
    }
    table_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 f85c2ef78a8e8d4ed2865df7424dcb5f59065067..76040a76011223b4ce2e96fae16fcfeb78879050 100644 (file)
--- a/src/test/regress/expected/foreign_key.out
+++ b/src/test/regress/expected/foreign_key.out
@@ -1921,7 +1921,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 c3a7cfcc019819820828ff7e42fb70cba96b225e..9ed1166c6666b8832496ddee2206ea900f12345c 100644 (file)
--- a/src/test/regress/sql/foreign_key.sql
+++ b/src/test/regress/sql/foreign_key.sql
@@ -1375,6 +1375,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