Repair problems occurring when multiple RI updates have to be done to the same

row within one query: we were firing check triggers before all the updates
were done, leading to bogus failures.  Fix by making the triggers queued by
an RI update go at the end of the outer query's trigger event list, thereby
effectively making the processing "breadth-first".  This was indeed how it
worked pre-8.0, so the bug does not occur in the 7.x branches.
Per report from Pavel Stehule.

3 files changed, 57 insertions, 39 deletions

diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index ac283605cfe..47de3ca6fc2 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/trigger.c,v 1.210.2.3 2007/07/17 17:45:40 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/trigger.c,v 1.210.2.4 2007/08/15 19:15:55 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -2284,6 +2284,7 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
 	AfterTriggerEvent event,
 				prev_event;
 	MemoryContext per_tuple_context;
+	bool		locally_opened = false;
 	Relation	rel = NULL;
 	TriggerDesc *trigdesc = NULL;
 	FmgrInfo   *finfo = NULL;
@@ -2316,6 +2317,19 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
 			 */
 			if (rel == NULL || rel->rd_id != event->ate_relid)
 			{
+				if (locally_opened)
+				{
+					/* close prior rel if any */
+					if (rel)
+						heap_close(rel, NoLock);
+					if (trigdesc)
+						FreeTriggerDesc(trigdesc);
+					if (finfo)
+						pfree(finfo);
+					Assert(instr == NULL);		/* never used in this case */
+				}
+				locally_opened = true;
+
 				if (estate)
 				{
 					/* Find target relation among estate's result rels */
@@ -2327,28 +2341,22 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
 					while (nr > 0)
 					{
 						if (rInfo->ri_RelationDesc->rd_id == event->ate_relid)
+						{
+							rel = rInfo->ri_RelationDesc;
+							trigdesc = rInfo->ri_TrigDesc;
+							finfo = rInfo->ri_TrigFunctions;
+							instr = rInfo->ri_TrigInstrument;
+							locally_opened = false;
 							break;
+						}
 						rInfo++;
 						nr--;
 					}
-					if (nr <= 0)	/* should not happen */
-						elog(ERROR, "could not find relation %u among query result relations",
-							 event->ate_relid);
-					rel = rInfo->ri_RelationDesc;
-					trigdesc = rInfo->ri_TrigDesc;
-					finfo = rInfo->ri_TrigFunctions;
-					instr = rInfo->ri_TrigInstrument;
 				}
-				else
+
+				if (locally_opened)
 				{
-					/* Hard way: we manage the resources for ourselves */
-					if (rel)
-						heap_close(rel, NoLock);
-					if (trigdesc)
-						FreeTriggerDesc(trigdesc);
-					if (finfo)
-						pfree(finfo);
-					Assert(instr == NULL);		/* never used in this case */
+					/* Hard way: open target relation for ourselves */
 
 					/*
 					 * We assume that an appropriate lock is still held by the
@@ -2373,6 +2381,7 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
 						palloc0(trigdesc->numtriggers * sizeof(FmgrInfo));
 
 					/* Never any EXPLAIN info in this case */
+					instr = NULL;
 				}
 			}
 
@@ -2423,7 +2432,7 @@ afterTriggerInvokeEvents(AfterTriggerEventList *events,
 	events->tail = prev_event;
 
 	/* Release working resources */
-	if (!estate)
+	if (locally_opened)
 	{
 		if (rel)
 			heap_close(rel, NoLock);
@@ -2552,11 +2561,13 @@ AfterTriggerEndQuery(EState *estate)
 	 * IMMEDIATE: all events we have decided to defer will be available for it
 	 * to fire.
 	 *
+	 * We loop in case a trigger queues more events.
+	 *
 	 * If we find no firable events, we don't have to increment
 	 * firing_counter.
 	 */
 	events = &afterTriggers->query_stack[afterTriggers->query_depth];
-	if (afterTriggerMarkEvents(events, &afterTriggers->events, true))
+	while (afterTriggerMarkEvents(events, &afterTriggers->events, true))
 	{
 		CommandId	firing_id = afterTriggers->firing_counter++;
 
@@ -2600,7 +2611,7 @@ AfterTriggerFireDeferred(void)
 		ActiveSnapshot = CopySnapshot(GetTransactionSnapshot());
 
 	/*
-	 * Run all the remaining triggers.	Loop until they are all gone, just in
+	 * Run all the remaining triggers.	Loop until they are all gone, in
 	 * case some trigger queues more for us to do.
 	 */
 	while (afterTriggerMarkEvents(events, NULL, false))
@@ -3163,7 +3174,7 @@ AfterTriggerSetState(ConstraintsSetStmt *stmt)
 	{
 		AfterTriggerEventList *events = &afterTriggers->events;
 
-		if (afterTriggerMarkEvents(events, NULL, true))
+		while (afterTriggerMarkEvents(events, NULL, true))
 		{
 			CommandId	firing_id = afterTriggers->firing_counter++;
 
diff --git a/src/backend/executor/spi.c b/src/backend/executor/spi.c
index 07f312e8a26..7d4f84f5ad8 100644
--- a/src/backend/executor/spi.c
+++ b/src/backend/executor/spi.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/executor/spi.c,v 1.165.2.3 2007/03/17 03:15:47 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/executor/spi.c,v 1.165.2.4 2007/08/15 19:15:55 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -39,9 +39,9 @@ static void _SPI_prepare_plan(const char *src, _SPI_plan *plan);
 static int _SPI_execute_plan(_SPI_plan *plan,
 				  Datum *Values, const char *Nulls,
 				  Snapshot snapshot, Snapshot crosscheck_snapshot,
-				  bool read_only, long tcount);
+				  bool read_only, bool fire_triggers, long tcount);
 
-static int	_SPI_pquery(QueryDesc *queryDesc, long tcount);
+static int	_SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, long tcount);
 
 static void _SPI_error_callback(void *arg);
 
@@ -315,7 +315,7 @@ SPI_execute(const char *src, bool read_only, long tcount)
 
 	res = _SPI_execute_plan(&plan, NULL, NULL,
 							InvalidSnapshot, InvalidSnapshot,
-							read_only, tcount);
+							read_only, true, tcount);
 
 	_SPI_end_call(true);
 	return res;
@@ -348,7 +348,7 @@ SPI_execute_plan(void *plan, Datum *Values, const char *Nulls,
 	res = _SPI_execute_plan((_SPI_plan *) plan,
 							Values, Nulls,
 							InvalidSnapshot, InvalidSnapshot,
-							read_only, tcount);
+							read_only, true, tcount);
 
 	_SPI_end_call(true);
 	return res;
@@ -363,9 +363,12 @@ SPI_execp(void *plan, Datum *Values, const char *Nulls, long tcount)
 
 /*
  * SPI_execute_snapshot -- identical to SPI_execute_plan, except that we allow
- * the caller to specify exactly which snapshots to use.  This is currently
- * not documented in spi.sgml because it is only intended for use by RI
- * triggers.
+ * the caller to specify exactly which snapshots to use.  Also, the caller
+ * may specify that AFTER triggers should be queued as part of the outer
+ * query rather than being fired immediately at the end of the command.
+ *
+ * This is currently not documented in spi.sgml because it is only intended
+ * for use by RI triggers.
  *
  * Passing snapshot == InvalidSnapshot will select the normal behavior of
  * fetching a new snapshot for each query.
@@ -374,7 +377,7 @@ int
 SPI_execute_snapshot(void *plan,
 					 Datum *Values, const char *Nulls,
 					 Snapshot snapshot, Snapshot crosscheck_snapshot,
-					 bool read_only, long tcount)
+					 bool read_only, bool fire_triggers, long tcount)
 {
 	int			res;
 
@@ -391,7 +394,7 @@ SPI_execute_snapshot(void *plan,
 	res = _SPI_execute_plan((_SPI_plan *) plan,
 							Values, Nulls,
 							snapshot, crosscheck_snapshot,
-							read_only, tcount);
+							read_only, fire_triggers, tcount);
 
 	_SPI_end_call(true);
 	return res;
@@ -1350,12 +1353,14 @@ _SPI_prepare_plan(const char *src, _SPI_plan *plan)
  *		behavior of taking a new snapshot for each query.
  * crosscheck_snapshot: for RI use, all others pass InvalidSnapshot
  * read_only: TRUE for read-only execution (no CommandCounterIncrement)
+ * fire_triggers: TRUE to fire AFTER triggers at end of query (normal case);
+ *		FALSE means any AFTER triggers are postponed to end of outer query
  * tcount: execution tuple-count limit, or 0 for none
  */
 static int
 _SPI_execute_plan(_SPI_plan *plan, Datum *Values, const char *Nulls,
 				  Snapshot snapshot, Snapshot crosscheck_snapshot,
-				  bool read_only, long tcount)
+				  bool read_only, bool fire_triggers, long tcount)
 {
 	volatile int my_res = 0;
 	volatile uint32 my_processed = 0;
@@ -1507,7 +1512,7 @@ _SPI_execute_plan(_SPI_plan *plan, Datum *Values, const char *Nulls,
 											crosscheck_snapshot,
 											dest,
 											paramLI, false);
-					res = _SPI_pquery(qdesc,
+					res = _SPI_pquery(qdesc, fire_triggers,
 									  queryTree->canSetTag ? tcount : 0);
 					FreeQueryDesc(qdesc);
 				}
@@ -1577,7 +1582,7 @@ fail:
 }
 
 static int
-_SPI_pquery(QueryDesc *queryDesc, long tcount)
+_SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, long tcount)
 {
 	int			operation = queryDesc->operation;
 	int			res;
@@ -1622,7 +1627,8 @@ _SPI_pquery(QueryDesc *queryDesc, long tcount)
 		ResetUsage();
 #endif
 
-	AfterTriggerBeginQuery();
+	if (fire_triggers)
+		AfterTriggerBeginQuery();
 
 	ExecutorStart(queryDesc, 0);
 
@@ -1639,7 +1645,8 @@ _SPI_pquery(QueryDesc *queryDesc, long tcount)
 	}
 
 	/* Take care of any queued AFTER triggers */
-	AfterTriggerEndQuery(queryDesc->estate);
+	if (fire_triggers)
+		AfterTriggerEndQuery(queryDesc->estate);
 
 	ExecutorEnd(queryDesc);
 
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index 747ef66bf77..af2955183cd 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -17,7 +17,7 @@
  *
  * Portions Copyright (c) 1996-2006, PostgreSQL Global Development Group
  *
- * $PostgreSQL: pgsql/src/backend/utils/adt/ri_triggers.c,v 1.89 2006/10/04 00:29:59 momjian Exp $
+ * $PostgreSQL: pgsql/src/backend/utils/adt/ri_triggers.c,v 1.89.2.1 2007/08/15 19:15:55 tgl Exp $
  *
  * ----------
  */
@@ -2717,7 +2717,7 @@ RI_Initial_Check(FkConstraint *fkconstraint, Relation rel, Relation pkrel)
 									  NULL, NULL,
 									  CopySnapshot(GetLatestSnapshot()),
 									  InvalidSnapshot,
-									  true, 1);
+									  true, false, 1);
 
 	/* Check result */
 	if (spi_result != SPI_OK_SELECT)
@@ -3146,7 +3146,7 @@ ri_PerformCheck(RI_QueryKey *qkey, void *qplan,
 	spi_result = SPI_execute_snapshot(qplan,
 									  vals, nulls,
 									  test_snapshot, crosscheck_snapshot,
-									  false, limit);
+									  false, false, limit);
 
 	/* Restore UID */
 	SetUserId(save_uid);