Allow FOR EACH ROW triggers on partitioned tables

Previously, FOR EACH ROW triggers were not allowed in partitioned
tables.  Now we allow AFTER triggers on them, and on trigger creation we
cascade to create an identical trigger in each partition.  We also clone
the triggers to each partition that is created or attached later.

This means that deferred unique keys are allowed on partitioned tables,
too.

Author: Álvaro Herrera
Reviewed-by: Peter Eisentraut, Simon Riggs, Amit Langote, Robert Haas,
	Thomas Munro
Discussion: https://postgr.es/m/20171229225319.ajltgss2ojkfd3kp@alvherre.pgsql

1 files changed, 284 insertions, 29 deletions

diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index fbd176b5d03..9d8df5986ec 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -20,6 +20,7 @@
 #include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
+#include "catalog/index.h"
 #include "catalog/indexing.h"
 #include "catalog/objectaccess.h"
 #include "catalog/pg_constraint.h"
@@ -123,7 +124,20 @@ static bool before_stmt_triggers_fired(Oid relid, CmdType cmdType);
  * TRIGGER, we build a pg_constraint entry internally.)
  *
  * indexOid, if nonzero, is the OID of an index associated with the constraint.
- * We do nothing with this except store it into pg_trigger.tgconstrindid.
+ * We do nothing with this except store it into pg_trigger.tgconstrindid;
+ * but when creating a trigger for a deferrable unique constraint on a
+ * partitioned table, its children are looked up.  Note we don't cope with
+ * invalid indexes in that case.
+ *
+ * funcoid, if nonzero, is the OID of the function to invoke.  When this is
+ * given, stmt->funcname is ignored.
+ *
+ * parentTriggerOid, if nonzero, is a trigger that begets this one; so that
+ * if that trigger is dropped, this one should be too.  (This is passed as
+ * Invalid by most callers; it's set here when recursing on a partition.)
+ *
+ * If whenClause is passed, it is an already-transformed expression for
+ * WHEN.  In this case, we ignore any that may come in stmt->whenClause.
  *
  * If isInternal is true then this is an internally-generated trigger.
  * This argument sets the tgisinternal field of the pg_trigger entry, and
@@ -133,6 +147,10 @@ static bool before_stmt_triggers_fired(Oid relid, CmdType cmdType);
  * relation, as well as ACL_EXECUTE on the trigger function.  For internal
  * triggers the caller must apply any required permission checks.
  *
+ * When called on partitioned tables, this function recurses to create the
+ * trigger on all the partitions, except if isInternal is true, in which
+ * case caller is expected to execute recursion on its own.
+ *
  * Note: can return InvalidObjectAddress if we decided to not create a trigger
  * at all, but a foreign-key constraint.  This is a kluge for backwards
  * compatibility.
@@ -140,13 +158,13 @@ static bool before_stmt_triggers_fired(Oid relid, CmdType cmdType);
 ObjectAddress
 CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 			  Oid relOid, Oid refRelOid, Oid constraintOid, Oid indexOid,
-			  bool isInternal)
+			  Oid funcoid, Oid parentTriggerOid, Node *whenClause,
+			  bool isInternal, bool in_partition)
 {
 	int16		tgtype;
 	int			ncolumns;
 	int16	   *columns;
 	int2vector *tgattr;
-	Node	   *whenClause;
 	List	   *whenRtable;
 	char	   *qual;
 	Datum		values[Natts_pg_trigger];
@@ -159,7 +177,6 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 	Relation	pgrel;
 	HeapTuple	tuple;
 	Oid			fargtypes[1];	/* dummy */
-	Oid			funcoid;
 	Oid			funcrettype;
 	Oid			trigoid;
 	char		internaltrigname[NAMEDATALEN];
@@ -169,6 +186,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 				referenced;
 	char	   *oldtablename = NULL;
 	char	   *newtablename = NULL;
+	bool		partition_recurse;
 
 	if (OidIsValid(relOid))
 		rel = heap_open(relOid, ShareRowExclusiveLock);
@@ -179,8 +197,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 	 * Triggers must be on tables or views, and there are additional
 	 * relation-type-specific restrictions.
 	 */
-	if (rel->rd_rel->relkind == RELKIND_RELATION ||
-		rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+	if (rel->rd_rel->relkind == RELKIND_RELATION)
 	{
 		/* Tables can't have INSTEAD OF triggers */
 		if (stmt->timing != TRIGGER_TYPE_BEFORE &&
@@ -190,13 +207,53 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 					 errmsg("\"%s\" is a table",
 							RelationGetRelationName(rel)),
 					 errdetail("Tables cannot have INSTEAD OF triggers.")));
-		/* Disallow ROW triggers on partitioned tables */
-		if (stmt->row && rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+	}
+	else if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+	{
+		/* Partitioned tables can't have INSTEAD OF triggers */
+		if (stmt->timing != TRIGGER_TYPE_BEFORE &&
+			stmt->timing != TRIGGER_TYPE_AFTER)
 			ereport(ERROR,
 					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
-					 errmsg("\"%s\" is a partitioned table",
+					 errmsg("\"%s\" is a table",
 							RelationGetRelationName(rel)),
-					 errdetail("Partitioned tables cannot have ROW triggers.")));
+					 errdetail("Tables cannot have INSTEAD OF triggers.")));
+
+		/*
+		 * FOR EACH ROW triggers have further restrictions
+		 */
+		if (stmt->row)
+		{
+			/*
+			 * BEFORE triggers FOR EACH ROW are forbidden, because they would
+			 * allow the user to direct the row to another partition, which
+			 * isn't implemented in the executor.
+			 */
+			if (stmt->timing != TRIGGER_TYPE_AFTER)
+				ereport(ERROR,
+						(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+						 errmsg("\"%s\" is a partitioned table",
+								RelationGetRelationName(rel)),
+						 errdetail("Partitioned tables cannot have BEFORE / FOR EACH ROW triggers.")));
+
+			/*
+			 * Disallow use of transition tables.
+			 *
+			 * Note that we have another restriction about transition tables
+			 * in partitions; search for 'has_superclass' below for an
+			 * explanation.  The check here is just to protect from the fact
+			 * that if we allowed it here, the creation would succeed for a
+			 * partitioned table with no partitions, but would be blocked by
+			 * the other restriction when the first partition was created,
+			 * which is very unfriendly behavior.
+			 */
+			if (stmt->transitionRels != NIL)
+				ereport(ERROR,
+						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+						 errmsg("\"%s\" is a partitioned table",
+								RelationGetRelationName(rel)),
+						 errdetail("Triggers on partitioned tables cannot have transition tables.")));
+		}
 	}
 	else if (rel->rd_rel->relkind == RELKIND_VIEW)
 	{
@@ -297,6 +354,18 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 		}
 	}
 
+	/*
+	 * When called on a partitioned table to create a FOR EACH ROW trigger
+	 * that's not internal, we create one trigger for each partition, too.
+	 *
+	 * For that, we'd better hold lock on all of them ahead of time.
+	 */
+	partition_recurse = !isInternal && stmt->row &&
+		rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE;
+	if (partition_recurse)
+		list_free(find_all_inheritors(RelationGetRelid(rel),
+									  ShareRowExclusiveLock, NULL));
+
 	/* Compute tgtype */
 	TRIGGER_CLEAR_TYPE(tgtype);
 	if (stmt->row)
@@ -484,9 +553,14 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 	}
 
 	/*
-	 * Parse the WHEN clause, if any
+	 * Parse the WHEN clause, if any and we weren't passed an already
+	 * transformed one.
+	 *
+	 * Note that as a side effect, we fill whenRtable when parsing.  If we got
+	 * an already parsed clause, this does not occur, which is what we want --
+	 * no point in adding redundant dependencies below.
 	 */
-	if (stmt->whenClause)
+	if (!whenClause && stmt->whenClause)
 	{
 		ParseState *pstate;
 		RangeTblEntry *rte;
@@ -577,17 +651,23 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 
 		free_parsestate(pstate);
 	}
-	else
+	else if (!whenClause)
 	{
 		whenClause = NULL;
 		whenRtable = NIL;
 		qual = NULL;
 	}
+	else
+	{
+		qual = nodeToString(whenClause);
+		whenRtable = NIL;
+	}
 
 	/*
 	 * Find and validate the trigger function.
 	 */
-	funcoid = LookupFuncName(stmt->funcname, 0, fargtypes, false);
+	if (!OidIsValid(funcoid))
+		funcoid = LookupFuncName(stmt->funcname, 0, fargtypes, false);
 	if (!isInternal)
 	{
 		aclresult = pg_proc_aclcheck(funcoid, GetUserId(), ACL_EXECUTE);
@@ -651,6 +731,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 											  stmt->deferrable,
 											  stmt->initdeferred,
 											  true,
+											  InvalidOid,	/* no parent */
 											  RelationGetRelid(rel),
 											  NULL, /* no conkey */
 											  0,
@@ -733,6 +814,11 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 
 	/*
 	 * Build the new pg_trigger tuple.
+	 *
+	 * When we're creating a trigger in a partition, we mark it as internal,
+	 * even though we don't do the isInternal magic in this function.  This
+	 * makes the triggers in partitions identical to the ones in the
+	 * partitioned tables, except that they are marked internal.
 	 */
 	memset(nulls, false, sizeof(nulls));
 
@@ -742,7 +828,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 	values[Anum_pg_trigger_tgfoid - 1] = ObjectIdGetDatum(funcoid);
 	values[Anum_pg_trigger_tgtype - 1] = Int16GetDatum(tgtype);
 	values[Anum_pg_trigger_tgenabled - 1] = CharGetDatum(TRIGGER_FIRES_ON_ORIGIN);
-	values[Anum_pg_trigger_tgisinternal - 1] = BoolGetDatum(isInternal);
+	values[Anum_pg_trigger_tgisinternal - 1] = BoolGetDatum(isInternal || in_partition);
 	values[Anum_pg_trigger_tgconstrrelid - 1] = ObjectIdGetDatum(constrrelid);
 	values[Anum_pg_trigger_tgconstrindid - 1] = ObjectIdGetDatum(indexOid);
 	values[Anum_pg_trigger_tgconstraint - 1] = ObjectIdGetDatum(constraintOid);
@@ -872,9 +958,8 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 		pfree(DatumGetPointer(values[Anum_pg_trigger_tgnewtable - 1]));
 
 	/*
-	 * Update relation's pg_class entry.  Crucial side-effect: other backends
-	 * (and this one too!) are sent SI message to make them rebuild relcache
-	 * entries.
+	 * Update relation's pg_class entry; if necessary; and if not, send an SI
+	 * message to make other backends (and this one) rebuild relcache entries.
 	 */
 	pgrel = heap_open(RelationRelationId, RowExclusiveLock);
 	tuple = SearchSysCacheCopy1(RELOID,
@@ -882,21 +967,21 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 	if (!HeapTupleIsValid(tuple))
 		elog(ERROR, "cache lookup failed for relation %u",
 			 RelationGetRelid(rel));
+	if (!((Form_pg_class) GETSTRUCT(tuple))->relhastriggers)
+	{
+		((Form_pg_class) GETSTRUCT(tuple))->relhastriggers = true;
 
-	((Form_pg_class) GETSTRUCT(tuple))->relhastriggers = true;
+		CatalogTupleUpdate(pgrel, &tuple->t_self, tuple);
 
-	CatalogTupleUpdate(pgrel, &tuple->t_self, tuple);
+		CommandCounterIncrement();
+	}
+	else
+		CacheInvalidateRelcacheByTuple(tuple);
 
 	heap_freetuple(tuple);
 	heap_close(pgrel, RowExclusiveLock);
 
 	/*
-	 * We used to try to update the rel's relcache entry here, but that's
-	 * fairly pointless since it will happen as a byproduct of the upcoming
-	 * CommandCounterIncrement...
-	 */
-
-	/*
 	 * Record dependencies for trigger.  Always place a normal dependency on
 	 * the function.
 	 */
@@ -928,11 +1013,18 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 		 * User CREATE TRIGGER, so place dependencies.  We make trigger be
 		 * auto-dropped if its relation is dropped or if the FK relation is
 		 * dropped.  (Auto drop is compatible with our pre-7.3 behavior.)
+		 *
+		 * Exception: if this trigger comes from a parent partitioned table,
+		 * then it's not separately drop-able, but goes away if the partition
+		 * does.
 		 */
 		referenced.classId = RelationRelationId;
 		referenced.objectId = RelationGetRelid(rel);
 		referenced.objectSubId = 0;
-		recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
+		recordDependencyOn(&myself, &referenced, OidIsValid(parentTriggerOid) ?
+						   DEPENDENCY_INTERNAL_AUTO :
+						   DEPENDENCY_AUTO);
+
 		if (OidIsValid(constrrelid))
 		{
 			referenced.classId = RelationRelationId;
@@ -954,6 +1046,13 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 			referenced.objectSubId = 0;
 			recordDependencyOn(&referenced, &myself, DEPENDENCY_INTERNAL);
 		}
+
+		/* Depends on the parent trigger, if there is one. */
+		if (OidIsValid(parentTriggerOid))
+		{
+			ObjectAddressSet(referenced, TriggerRelationId, parentTriggerOid);
+			recordDependencyOn(&myself, &referenced, DEPENDENCY_INTERNAL_AUTO);
+		}
 	}
 
 	/* If column-specific trigger, add normal dependencies on columns */
@@ -974,7 +1073,7 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 	 * If it has a WHEN clause, add dependencies on objects mentioned in the
 	 * expression (eg, functions, as well as any columns used).
 	 */
-	if (whenClause != NULL)
+	if (whenRtable != NIL)
 		recordDependencyOnExpr(&myself, whenClause, whenRtable,
 							   DEPENDENCY_NORMAL);
 
@@ -982,6 +1081,112 @@ CreateTrigger(CreateTrigStmt *stmt, const char *queryString,
 	InvokeObjectPostCreateHookArg(TriggerRelationId, trigoid, 0,
 								  isInternal);
 
+	/*
+	 * Lastly, create the trigger on child relations, if needed.
+	 */
+	if (partition_recurse)
+	{
+		PartitionDesc partdesc = RelationGetPartitionDesc(rel);
+		List	   *idxs = NIL;
+		List	   *childTbls = NIL;
+		ListCell   *l;
+		int			i;
+		MemoryContext oldcxt,
+					perChildCxt;
+
+		perChildCxt = AllocSetContextCreate(CurrentMemoryContext,
+											"part trig clone",
+											ALLOCSET_SMALL_SIZES);
+
+		/*
+		 * When a trigger is being created associated with an index, we'll
+		 * need to associate the trigger in each child partition with the
+		 * corresponding index on it.
+		 */
+		if (OidIsValid(indexOid))
+		{
+			ListCell   *l;
+			List	   *idxs = NIL;
+
+			idxs = find_inheritance_children(indexOid, ShareRowExclusiveLock);
+			foreach(l, idxs)
+				childTbls = lappend_oid(childTbls,
+										IndexGetRelation(lfirst_oid(l),
+														 false));
+		}
+
+		oldcxt = MemoryContextSwitchTo(perChildCxt);
+
+		/* Iterate to create the trigger on each existing partition */
+		for (i = 0; i < partdesc->nparts; i++)
+		{
+			Oid			indexOnChild = InvalidOid;
+			ListCell   *l2;
+			CreateTrigStmt *childStmt;
+			Relation	childTbl;
+			Node	   *qual;
+			bool		found_whole_row;
+
+			childTbl = heap_open(partdesc->oids[i], ShareRowExclusiveLock);
+
+			/* Find which of the child indexes is the one on this partition */
+			if (OidIsValid(indexOid))
+			{
+				forboth(l, idxs, l2, childTbls)
+				{
+					if (lfirst_oid(l2) == partdesc->oids[i])
+					{
+						indexOnChild = lfirst_oid(l);
+						break;
+					}
+				}
+				if (!OidIsValid(indexOnChild))
+					elog(ERROR, "failed to find index matching index \"%s\" in partition \"%s\"",
+						 get_rel_name(indexOid),
+						 get_rel_name(partdesc->oids[i]));
+			}
+
+			/*
+			 * Initialize our fabricated parse node by copying the original
+			 * one, then resetting fields that we pass separately.
+			 */
+			childStmt = (CreateTrigStmt *) copyObject(stmt);
+			childStmt->funcname = NIL;
+			childStmt->args = NIL;
+			childStmt->whenClause = NULL;
+
+			/* If there is a WHEN clause, create a modified copy of it */
+			qual = copyObject(whenClause);
+			qual = (Node *)
+				map_partition_varattnos((List *) qual, PRS2_OLD_VARNO,
+										childTbl, rel,
+										&found_whole_row);
+			if (found_whole_row)
+				elog(ERROR, "unexpected whole-row reference found in trigger WHEN clause");
+			qual = (Node *)
+				map_partition_varattnos((List *) qual, PRS2_NEW_VARNO,
+										childTbl, rel,
+										&found_whole_row);
+			if (found_whole_row)
+				elog(ERROR, "unexpected whole-row reference found in trigger WHEN clause");
+
+			CreateTrigger(childStmt, queryString,
+						  partdesc->oids[i], refRelOid,
+						  InvalidOid, indexOnChild,
+						  funcoid, trigoid, qual,
+						  isInternal, true);
+
+			heap_close(childTbl, NoLock);
+
+			MemoryContextReset(perChildCxt);
+		}
+
+		MemoryContextSwitchTo(oldcxt);
+		MemoryContextDelete(perChildCxt);
+		list_free(idxs);
+		list_free(childTbls);
+	}
+
 	/* Keep lock on target rel until end of xact */
 	heap_close(rel, NoLock);
 
@@ -1579,7 +1784,7 @@ renametrig(RenameStmt *stmt)
  */
 void
 EnableDisableTrigger(Relation rel, const char *tgname,
-					 char fires_when, bool skip_system)
+					 char fires_when, bool skip_system, LOCKMODE lockmode)
 {
 	Relation	tgrel;
 	int			nkeys;
@@ -1642,6 +1847,27 @@ EnableDisableTrigger(Relation rel, const char *tgname,
 
 			heap_freetuple(newtup);
 
+			/*
+			 * When altering FOR EACH ROW triggers on a partitioned table, do
+			 * the same on the partitions as well.
+			 */
+			if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE &&
+				(TRIGGER_FOR_ROW(oldtrig->tgtype)))
+			{
+				PartitionDesc partdesc = RelationGetPartitionDesc(rel);
+				int			i;
+
+				for (i = 0; i < partdesc->nparts; i++)
+				{
+					Relation	part;
+
+					part = relation_open(partdesc->oids[i], lockmode);
+					EnableDisableTrigger(part, NameStr(oldtrig->tgname),
+										 fires_when, skip_system, lockmode);
+					heap_close(part, NoLock);	/* keep lock till commit */
+				}
+			}
+
 			changed = true;
 		}
 
@@ -5123,6 +5349,9 @@ AfterTriggerSetState(ConstraintsSetStmt *stmt)
 		 * constraints within the first search-path schema that has any
 		 * matches, but disregard matches in schemas beyond the first match.
 		 * (This is a bit odd but it's the historical behavior.)
+		 *
+		 * A constraint in a partitioned table may have corresponding
+		 * constraints in the partitions.  Grab those too.
 		 */
 		conrel = heap_open(ConstraintRelationId, AccessShareLock);
 
@@ -5217,6 +5446,32 @@ AfterTriggerSetState(ConstraintsSetStmt *stmt)
 								constraint->relname)));
 		}
 
+		/*
+		 * Scan for any possible descendants of the constraints.  We append
+		 * whatever we find to the same list that we're scanning; this has the
+		 * effect that we create new scans for those, too, so if there are
+		 * further descendents, we'll also catch them.
+		 */
+		foreach(lc, conoidlist)
+		{
+			Oid			parent = lfirst_oid(lc);
+			ScanKeyData key;
+			SysScanDesc scan;
+			HeapTuple	tuple;
+
+			ScanKeyInit(&key,
+						Anum_pg_constraint_conparentid,
+						BTEqualStrategyNumber, F_OIDEQ,
+						ObjectIdGetDatum(parent));
+
+			scan = systable_beginscan(conrel, ConstraintParentIndexId, true, NULL, 1, &key);
+
+			while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+				conoidlist = lappend_oid(conoidlist, HeapTupleGetOid(tuple));
+
+			systable_endscan(scan);
+		}
+
 		heap_close(conrel, AccessShareLock);
 
 		/*