Repair RI trigger visibility problems (this time for sure ;-)) per recent

discussion on pgsql-hackers: in READ COMMITTED mode we just have to force
a QuerySnapshot update in the trigger, but in SERIALIZABLE mode we have
to run the scan under a current snapshot and then complain if any rows
would be updated/deleted that are not visible in the transaction snapshot.

5 files changed, 49 insertions, 34 deletions

diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 6b92920bcd1..bfdc94c6d51 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -26,7 +26,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/executor/execMain.c,v 1.219 2003/09/25 18:58:35 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/executor/execMain.c,v 1.220 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -104,8 +104,14 @@ static void EvalPlanQualStop(evalPlanQual *epq);
  * field of the QueryDesc is filled in to describe the tuples that will be
  * returned, and the internal fields (estate and planstate) are set up.
  *
- * If useSnapshotNow is true, run the query with SnapshotNow time qual rules
- * instead of the normal use of QuerySnapshot.
+ * If useCurrentSnapshot is true, run the query with the latest available
+ * snapshot, instead of the normal QuerySnapshot.  Also, if it's an update
+ * or delete query, check that the rows to be updated or deleted would be
+ * visible to the normal QuerySnapshot.  (This is a special-case behavior
+ * needed for referential integrity updates in serializable transactions.
+ * We must check all currently-committed rows, but we want to throw a
+ * can't-serialize error if any rows that would need updates would not be
+ * visible under the normal serializable snapshot.)
  *
  * If explainOnly is true, we are not actually intending to run the plan,
  * only to set up for EXPLAIN; so skip unwanted side-effects.
@@ -115,7 +121,7 @@ static void EvalPlanQualStop(evalPlanQual *epq);
  * ----------------------------------------------------------------
  */
 void
-ExecutorStart(QueryDesc *queryDesc, bool useSnapshotNow, bool explainOnly)
+ExecutorStart(QueryDesc *queryDesc, bool useCurrentSnapshot, bool explainOnly)
 {
 	EState	   *estate;
 	MemoryContext oldcontext;
@@ -157,15 +163,18 @@ ExecutorStart(QueryDesc *queryDesc, bool useSnapshotNow, bool explainOnly)
 	 * the life of this query, even if it outlives the current command and
 	 * current snapshot.
 	 */
-	if (useSnapshotNow)
+	if (useCurrentSnapshot)
 	{
-		estate->es_snapshot = SnapshotNow;
-		estate->es_snapshot_cid = GetCurrentCommandId();
+		/* RI update/delete query --- must use an up-to-date snapshot */
+		estate->es_snapshot = CopyCurrentSnapshot();
+		/* crosscheck updates/deletes against transaction snapshot */
+		estate->es_crosscheck_snapshot = CopyQuerySnapshot();
 	}
 	else
 	{
+		/* normal query --- use query snapshot, no crosscheck */
 		estate->es_snapshot = CopyQuerySnapshot();
-		estate->es_snapshot_cid = estate->es_snapshot->curcid;
+		estate->es_crosscheck_snapshot = SnapshotAny;
 	}
 
 	/*
@@ -1118,7 +1127,7 @@ lnext:	;
 
 					tuple.t_self = *((ItemPointer) DatumGetPointer(datum));
 					test = heap_mark4update(erm->relation, &tuple, &buffer,
-											estate->es_snapshot_cid);
+											estate->es_snapshot->curcid);
 					ReleaseBuffer(buffer);
 					switch (test)
 					{
@@ -1278,7 +1287,7 @@ ExecSelect(TupleTableSlot *slot,
 	if (estate->es_into_relation_descriptor != NULL)
 	{
 		heap_insert(estate->es_into_relation_descriptor, tuple,
-					estate->es_snapshot_cid);
+					estate->es_snapshot->curcid);
 		IncrAppended();
 	}
 
@@ -1354,7 +1363,7 @@ ExecInsert(TupleTableSlot *slot,
 	 * insert the tuple
 	 */
 	newId = heap_insert(resultRelationDesc, tuple,
-						estate->es_snapshot_cid);
+						estate->es_snapshot->curcid);
 
 	IncrAppended();
 	(estate->es_processed)++;
@@ -1406,7 +1415,7 @@ ExecDelete(TupleTableSlot *slot,
 		bool		dodelete;
 
 		dodelete = ExecBRDeleteTriggers(estate, resultRelInfo, tupleid,
-										estate->es_snapshot_cid);
+										estate->es_snapshot->curcid);
 
 		if (!dodelete)			/* "do nothing" */
 			return;
@@ -1418,7 +1427,8 @@ ExecDelete(TupleTableSlot *slot,
 ldelete:;
 	result = heap_delete(resultRelationDesc, tupleid,
 						 &ctid,
-						 estate->es_snapshot_cid,
+						 estate->es_snapshot->curcid,
+						 estate->es_crosscheck_snapshot,
 						 true /* wait for commit */);
 	switch (result)
 	{
@@ -1517,7 +1527,7 @@ ExecUpdate(TupleTableSlot *slot,
 
 		newtuple = ExecBRUpdateTriggers(estate, resultRelInfo,
 										tupleid, tuple,
-										estate->es_snapshot_cid);
+										estate->es_snapshot->curcid);
 
 		if (newtuple == NULL)	/* "do nothing" */
 			return;
@@ -1553,7 +1563,8 @@ lreplace:;
 	 */
 	result = heap_update(resultRelationDesc, tupleid, tuple,
 						 &ctid,
-						 estate->es_snapshot_cid,
+						 estate->es_snapshot->curcid,
+						 estate->es_crosscheck_snapshot,
 						 true /* wait for commit */);
 	switch (result)
 	{
@@ -2039,7 +2050,7 @@ EvalPlanQualStart(evalPlanQual *epq, EState *estate, evalPlanQual *priorepq)
 	 */
 	epqstate->es_direction = ForwardScanDirection;
 	epqstate->es_snapshot = estate->es_snapshot;
-	epqstate->es_snapshot_cid = estate->es_snapshot_cid;
+	epqstate->es_crosscheck_snapshot = estate->es_crosscheck_snapshot;
 	epqstate->es_range_table = estate->es_range_table;
 	epqstate->es_result_relations = estate->es_result_relations;
 	epqstate->es_num_result_relations = estate->es_num_result_relations;
diff --git a/src/backend/executor/execUtils.c b/src/backend/executor/execUtils.c
index c9c7ef79396..1ee99cb359e 100644
--- a/src/backend/executor/execUtils.c
+++ b/src/backend/executor/execUtils.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/executor/execUtils.c,v 1.105 2003/09/25 18:58:35 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/executor/execUtils.c,v 1.106 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -178,7 +178,7 @@ CreateExecutorState(void)
 	 */
 	estate->es_direction = ForwardScanDirection;
 	estate->es_snapshot = SnapshotNow;
-	estate->es_snapshot_cid = FirstCommandId;
+	estate->es_crosscheck_snapshot = SnapshotAny; /* means no crosscheck */
 	estate->es_range_table = NIL;
 
 	estate->es_result_relations = NULL;
diff --git a/src/backend/executor/nodeSubplan.c b/src/backend/executor/nodeSubplan.c
index 488a37b24d3..971dd5879db 100644
--- a/src/backend/executor/nodeSubplan.c
+++ b/src/backend/executor/nodeSubplan.c
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/executor/nodeSubplan.c,v 1.57 2003/09/25 18:58:35 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/executor/nodeSubplan.c,v 1.58 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -709,7 +709,7 @@ ExecInitSubPlan(SubPlanState *node, EState *estate)
 	sp_estate->es_tupleTable =
 		ExecCreateTupleTable(ExecCountSlotsNode(subplan->plan) + 10);
 	sp_estate->es_snapshot = estate->es_snapshot;
-	sp_estate->es_snapshot_cid = estate->es_snapshot_cid;
+	sp_estate->es_crosscheck_snapshot = estate->es_crosscheck_snapshot;
 	sp_estate->es_instrument = estate->es_instrument;
 
 	/*
diff --git a/src/backend/executor/nodeSubqueryscan.c b/src/backend/executor/nodeSubqueryscan.c
index f8d2640349f..1c15f5ff391 100644
--- a/src/backend/executor/nodeSubqueryscan.c
+++ b/src/backend/executor/nodeSubqueryscan.c
@@ -12,7 +12,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/executor/nodeSubqueryscan.c,v 1.21 2003/09/25 18:58:35 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/executor/nodeSubqueryscan.c,v 1.22 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -177,7 +177,7 @@ ExecInitSubqueryScan(SubqueryScan *node, EState *estate)
 	sp_estate->es_tupleTable =
 		ExecCreateTupleTable(ExecCountSlotsNode(node->subplan) + 10);
 	sp_estate->es_snapshot = estate->es_snapshot;
-	sp_estate->es_snapshot_cid = estate->es_snapshot_cid;
+	sp_estate->es_crosscheck_snapshot = estate->es_crosscheck_snapshot;
 	sp_estate->es_instrument = estate->es_instrument;
 
 	/*
diff --git a/src/backend/executor/spi.c b/src/backend/executor/spi.c
index 2626b728e94..e9e2084fdaf 100644
--- a/src/backend/executor/spi.c
+++ b/src/backend/executor/spi.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/executor/spi.c,v 1.106 2003/09/25 18:58:35 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/executor/spi.c,v 1.107 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -33,11 +33,11 @@ static int	_SPI_curid = -1;
 
 static int	_SPI_execute(const char *src, int tcount, _SPI_plan *plan);
 static int	_SPI_pquery(QueryDesc *queryDesc, bool runit,
-						bool useSnapshotNow, int tcount);
+						bool useCurrentSnapshot, int tcount);
 
 static int _SPI_execute_plan(_SPI_plan *plan,
 							 Datum *Values, const char *Nulls,
-							 bool useSnapshotNow, int tcount);
+							 bool useCurrentSnapshot, int tcount);
 
 static void _SPI_cursor_operation(Portal portal, bool forward, int count,
 					  DestReceiver *dest);
@@ -245,12 +245,14 @@ SPI_execp(void *plan, Datum *Values, const char *Nulls, int tcount)
 }
 
 /*
- * SPI_execp_now -- identical to SPI_execp, except that we use SnapshotNow
- * instead of the normal QuerySnapshot.  This is currently not documented
- * in spi.sgml because it is only intended for use by RI triggers.
+ * SPI_execp_current -- identical to SPI_execp, except that we expose the
+ * Executor option to use a current snapshot instead of the normal
+ * QuerySnapshot.  This is currently not documented in spi.sgml because
+ * it is only intended for use by RI triggers.
  */
 int
-SPI_execp_now(void *plan, Datum *Values, const char *Nulls, int tcount)
+SPI_execp_current(void *plan, Datum *Values, const char *Nulls,
+				  bool useCurrentSnapshot, int tcount)
 {
 	int			res;
 
@@ -264,7 +266,8 @@ SPI_execp_now(void *plan, Datum *Values, const char *Nulls, int tcount)
 	if (res < 0)
 		return res;
 
-	res = _SPI_execute_plan((_SPI_plan *) plan, Values, Nulls, true, tcount);
+	res = _SPI_execute_plan((_SPI_plan *) plan, Values, Nulls,
+							useCurrentSnapshot, tcount);
 
 	_SPI_end_call(true);
 	return res;
@@ -1124,7 +1127,7 @@ _SPI_execute(const char *src, int tcount, _SPI_plan *plan)
 
 static int
 _SPI_execute_plan(_SPI_plan *plan, Datum *Values, const char *Nulls,
-				  bool useSnapshotNow, int tcount)
+				  bool useCurrentSnapshot, int tcount)
 {
 	List	   *query_list_list = plan->qtlist;
 	List	   *plan_list = plan->ptlist;
@@ -1195,7 +1198,7 @@ _SPI_execute_plan(_SPI_plan *plan, Datum *Values, const char *Nulls,
 			{
 				qdesc = CreateQueryDesc(queryTree, planTree, dest,
 										paramLI, false);
-				res = _SPI_pquery(qdesc, true, useSnapshotNow,
+				res = _SPI_pquery(qdesc, true, useCurrentSnapshot,
 								  queryTree->canSetTag ? tcount : 0);
 				if (res < 0)
 					return res;
@@ -1208,7 +1211,8 @@ _SPI_execute_plan(_SPI_plan *plan, Datum *Values, const char *Nulls,
 }
 
 static int
-_SPI_pquery(QueryDesc *queryDesc, bool runit, bool useSnapshotNow, int tcount)
+_SPI_pquery(QueryDesc *queryDesc, bool runit,
+			bool useCurrentSnapshot, int tcount)
 {
 	int			operation = queryDesc->operation;
 	int			res;
@@ -1245,7 +1249,7 @@ _SPI_pquery(QueryDesc *queryDesc, bool runit, bool useSnapshotNow, int tcount)
 		ResetUsage();
 #endif
 
-	ExecutorStart(queryDesc, useSnapshotNow, false);
+	ExecutorStart(queryDesc, useCurrentSnapshot, false);
 
 	ExecutorRun(queryDesc, ForwardScanDirection, (long) tcount);