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

Author: alvherre

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);

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

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