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.

src/backend/access/heap/heapam.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/access/heap/heapam.c,v 1.156 2003/09/25 06:57:56 petere Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/access/heap/heapam.c,v 1.157 2003/10/01 21:30:52 tgl Exp $
  *
  *
  * INTERFACE ROUTINES
@@ -1207,14 +1207,23 @@ simple_heap_insert(Relation relation, HeapTuple tup)
  * NB: do not call this directly unless you are prepared to deal with
  * concurrent-update conditions.  Use simple_heap_delete instead.
  *
+ *	relation - table to be modified
+ *	tid - TID of tuple to be deleted
+ *	ctid - output parameter, used only for failure case (see below)
+ *	cid - delete command ID to use in verifying tuple visibility
+ *	crosscheck - if not SnapshotAny, also check tuple against this
+ *	wait - true if should wait for any conflicting update to commit/abort
+ *
  * Normal, successful return value is HeapTupleMayBeUpdated, which
  * actually means we did delete it.  Failure return codes are
  * HeapTupleSelfUpdated, HeapTupleUpdated, or HeapTupleBeingUpdated
- * (the last only possible if wait == false).
+ * (the last only possible if wait == false).  On a failure return,
+ * *ctid is set to the ctid link of the target tuple (possibly a later
+ * version of the row).
  */
 int
 heap_delete(Relation relation, ItemPointer tid,
-			ItemPointer ctid, CommandId cid, bool wait)
+			ItemPointer ctid, CommandId cid, Snapshot crosscheck, bool wait)
 {
 	ItemId		lp;
 	HeapTupleData tp;
@@ -1240,7 +1249,7 @@ heap_delete(Relation relation, ItemPointer tid,
 	tp.t_tableOid = relation->rd_id;
 
 l1:
-	result = HeapTupleSatisfiesUpdate(&tp, cid);
+	result = HeapTupleSatisfiesUpdate(tp.t_data, cid);
 
 	if (result == HeapTupleInvisible)
 	{
@@ -1278,6 +1287,14 @@ l1:
 		else
 			result = HeapTupleUpdated;
 	}
+
+	if (crosscheck != SnapshotAny && result == HeapTupleMayBeUpdated)
+	{
+		/* Perform additional check for serializable RI updates */
+		if (!HeapTupleSatisfiesSnapshot(tp.t_data, crosscheck))
+			result = HeapTupleUpdated;
+	}
+
 	if (result != HeapTupleMayBeUpdated)
 	{
 		Assert(result == HeapTupleSelfUpdated ||
@@ -1378,7 +1395,7 @@ simple_heap_delete(Relation relation, ItemPointer tid)
 
 	result = heap_delete(relation, tid,
 						 &ctid,
-						 GetCurrentCommandId(),
+						 GetCurrentCommandId(), SnapshotAny,
 						 true /* wait for commit */);
 	switch (result)
 	{
@@ -1407,14 +1424,26 @@ simple_heap_delete(Relation relation, ItemPointer tid)
  * NB: do not call this directly unless you are prepared to deal with
  * concurrent-update conditions.  Use simple_heap_update instead.
  *
+ *	relation - table to be modified
+ *	otid - TID of old tuple to be replaced
+ *	newtup - newly constructed tuple data to store
+ *	ctid - output parameter, used only for failure case (see below)
+ *	cid - update command ID to use in verifying old tuple visibility
+ *	crosscheck - if not SnapshotAny, also check old tuple against this
+ *	wait - true if should wait for any conflicting update to commit/abort
+ *
  * Normal, successful return value is HeapTupleMayBeUpdated, which
  * actually means we *did* update it.  Failure return codes are
  * HeapTupleSelfUpdated, HeapTupleUpdated, or HeapTupleBeingUpdated
- * (the last only possible if wait == false).
+ * (the last only possible if wait == false).  On a failure return,
+ * *ctid is set to the ctid link of the old tuple (possibly a later
+ * version of the row).
+ * On success, newtup->t_self is set to the TID where the new tuple
+ * was inserted.
  */
 int
 heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
-			ItemPointer ctid, CommandId cid, bool wait)
+			ItemPointer ctid, CommandId cid, Snapshot crosscheck, bool wait)
 {
 	ItemId		lp;
 	HeapTupleData oldtup;
@@ -1450,7 +1479,7 @@ heap_update(Relation relation, ItemPointer otid, HeapTuple newtup,
 	 */
 
 l2:
-	result = HeapTupleSatisfiesUpdate(&oldtup, cid);
+	result = HeapTupleSatisfiesUpdate(oldtup.t_data, cid);
 
 	if (result == HeapTupleInvisible)
 	{
@@ -1488,6 +1517,14 @@ l2:
 		else
 			result = HeapTupleUpdated;
 	}
+
+	if (crosscheck != SnapshotAny && result == HeapTupleMayBeUpdated)
+	{
+		/* Perform additional check for serializable RI updates */
+		if (!HeapTupleSatisfiesSnapshot(oldtup.t_data, crosscheck))
+			result = HeapTupleUpdated;
+	}
+
 	if (result != HeapTupleMayBeUpdated)
 	{
 		Assert(result == HeapTupleSelfUpdated ||
@@ -1718,7 +1755,7 @@ simple_heap_update(Relation relation, ItemPointer otid, HeapTuple tup)
 
 	result = heap_update(relation, otid, tup,
 						 &ctid,
-						 GetCurrentCommandId(),
+						 GetCurrentCommandId(), SnapshotAny,
 						 true /* wait for commit */);
 	switch (result)
 	{
@@ -1767,7 +1804,7 @@ heap_mark4update(Relation relation, HeapTuple tuple, Buffer *buffer,
 	tuple->t_len = ItemIdGetLength(lp);
 
 l3:
-	result = HeapTupleSatisfiesUpdate(tuple, cid);
+	result = HeapTupleSatisfiesUpdate(tuple->t_data, cid);
 
 	if (result == HeapTupleInvisible)
 	{

src/backend/commands/async.c

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/async.c,v 1.100 2003/09/15 23:33:39 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/async.c,v 1.101 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -537,7 +537,7 @@ AtCommit_Notify(void)
 				 */
 				result = heap_update(lRel, &lTuple->t_self, rTuple,
 									 &ctid,
-									 GetCurrentCommandId(),
+									 GetCurrentCommandId(), SnapshotAny,
 									 false /* no wait for commit */);
 				switch (result)
 				{

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;

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;

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;
 
 	/*

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;
 
 	/*

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

src/backend/storage/ipc/sinval.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/sinval.c,v 1.60 2003/09/24 18:54:01 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/storage/ipc/sinval.c,v 1.61 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -330,10 +330,10 @@ GetSnapshotData(Snapshot snapshot, bool serializable)
 	 * lastBackend would be sufficient.  But it seems better to do the
 	 * malloc while not holding the lock, so we can't look at lastBackend.
 	 *
-	 * if (snapshot->xip != NULL) no need to free and reallocate xip;
-	 *
-	 * We can reuse the old xip array, because MaxBackends does not change at
-	 * runtime.
+	 * This does open a possibility for avoiding repeated malloc/free:
+	 * since MaxBackends does not change at runtime, we can simply reuse
+	 * the previous xip array if any.  (This relies on the fact that all
+	 * calls pass static SnapshotData structs.)
 	 */
 	if (snapshot->xip == NULL)
 	{

src/backend/utils/adt/ri_triggers.c

@@ -17,7 +17,7 @@
  *
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  *
- * $Header: /cvsroot/pgsql/src/backend/utils/adt/ri_triggers.c,v 1.60 2003/09/29 00:05:25 petere Exp $
+ * $Header: /cvsroot/pgsql/src/backend/utils/adt/ri_triggers.c,v 1.61 2003/10/01 21:30:52 tgl Exp $
  *
  * ----------
  */
@@ -157,6 +157,7 @@ static void *ri_PlanCheck(const char *querystr, int nargs, Oid *argtypes,
 static bool ri_PerformCheck(RI_QueryKey *qkey, void *qplan,
 				Relation fk_rel, Relation pk_rel,
 				HeapTuple old_tuple, HeapTuple new_tuple,
+				bool detectNewRows,
 				int expect_OK, const char *constrname);
 static void ri_ExtractValues(RI_QueryKey *qkey, int key_idx,
 				 Relation rel, HeapTuple tuple,
@@ -276,6 +277,7 @@ RI_FKey_check(PG_FUNCTION_ARGS)
 		ri_PerformCheck(&qkey, qplan,
 						fk_rel, pk_rel,
 						NULL, NULL,
+						false,
 						SPI_OK_SELECT,
 						tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -433,6 +435,7 @@ RI_FKey_check(PG_FUNCTION_ARGS)
 	ri_PerformCheck(&qkey, qplan,
 					fk_rel, pk_rel,
 					NULL, new_row,
+					false,
 					SPI_OK_SELECT,
 					tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -594,6 +597,7 @@ ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
 	result = ri_PerformCheck(&qkey, qplan,
 							 fk_rel, pk_rel,
 							 old_row, NULL,
+							 true,		/* treat like update */
 							 SPI_OK_SELECT, NULL);
 
 	if (SPI_finish() != SPI_OK_FINISH)
@@ -752,6 +756,7 @@ RI_FKey_noaction_del(PG_FUNCTION_ARGS)
 			ri_PerformCheck(&qkey, qplan,
 							fk_rel, pk_rel,
 							old_row, NULL,
+							true, /* must detect new rows */
 							SPI_OK_SELECT,
 							tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -942,6 +947,7 @@ RI_FKey_noaction_upd(PG_FUNCTION_ARGS)
 			ri_PerformCheck(&qkey, qplan,
 							fk_rel, pk_rel,
 							old_row, NULL,
+							true, /* must detect new rows */
 							SPI_OK_SELECT,
 							tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -1102,6 +1108,7 @@ RI_FKey_cascade_del(PG_FUNCTION_ARGS)
 			ri_PerformCheck(&qkey, qplan,
 							fk_rel, pk_rel,
 							old_row, NULL,
+							true, /* must detect new rows */
 							SPI_OK_DELETE,
 							tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -1285,6 +1292,7 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
 			ri_PerformCheck(&qkey, qplan,
 							fk_rel, pk_rel,
 							old_row, new_row,
+							true, /* must detect new rows */
 							SPI_OK_UPDATE,
 							tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -1453,6 +1461,7 @@ RI_FKey_restrict_del(PG_FUNCTION_ARGS)
 			ri_PerformCheck(&qkey, qplan,
 							fk_rel, pk_rel,
 							old_row, NULL,
+							true, /* must detect new rows */
 							SPI_OK_SELECT,
 							tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -1633,6 +1642,7 @@ RI_FKey_restrict_upd(PG_FUNCTION_ARGS)
 			ri_PerformCheck(&qkey, qplan,
 							fk_rel, pk_rel,
 							old_row, NULL,
+							true, /* must detect new rows */
 							SPI_OK_SELECT,
 							tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -1802,6 +1812,7 @@ RI_FKey_setnull_del(PG_FUNCTION_ARGS)
 			ri_PerformCheck(&qkey, qplan,
 							fk_rel, pk_rel,
 							old_row, NULL,
+							true, /* must detect new rows */
 							SPI_OK_UPDATE,
 							tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -2019,6 +2030,7 @@ RI_FKey_setnull_upd(PG_FUNCTION_ARGS)
 			ri_PerformCheck(&qkey, qplan,
 							fk_rel, pk_rel,
 							old_row, NULL,
+							true, /* must detect new rows */
 							SPI_OK_UPDATE,
 							tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -2188,6 +2200,7 @@ RI_FKey_setdefault_del(PG_FUNCTION_ARGS)
 			ri_PerformCheck(&qkey, qplan,
 							fk_rel, pk_rel,
 							old_row, NULL,
+							true, /* must detect new rows */
 							SPI_OK_UPDATE,
 							tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -2392,6 +2405,7 @@ RI_FKey_setdefault_upd(PG_FUNCTION_ARGS)
 			ri_PerformCheck(&qkey, qplan,
 							fk_rel, pk_rel,
 							old_row, NULL,
+							true, /* must detect new rows */
 							SPI_OK_UPDATE,
 							tgargs[RI_CONSTRAINT_NAME_ARGNO]);
 
@@ -2788,11 +2802,13 @@ static bool
 ri_PerformCheck(RI_QueryKey *qkey, void *qplan,
 				Relation fk_rel, Relation pk_rel,
 				HeapTuple old_tuple, HeapTuple new_tuple,
+				bool detectNewRows,
 				int expect_OK, const char *constrname)
 {
 	Relation	query_rel,
 				source_rel;
 	int			key_idx;
+	bool		useCurrentSnapshot;
 	int			limit;
 	int			spi_result;
 	AclId		save_uid;
@@ -2842,9 +2858,25 @@ ri_PerformCheck(RI_QueryKey *qkey, void *qplan,
 						 vals, nulls);
 	}
 
-	/* Switch to proper UID to perform check as */
-	save_uid = GetUserId();
-	SetUserId(RelationGetForm(query_rel)->relowner);
+	/*
+	 * In READ COMMITTED mode, we just need to make sure the regular query
+	 * snapshot is up-to-date, and we will see all rows that could be
+	 * interesting.  In SERIALIZABLE mode, we can't update the regular query
+	 * snapshot.  If the caller passes detectNewRows == false then it's okay
+	 * to do the query with the transaction snapshot; otherwise we tell the
+	 * executor to force a current snapshot (and error out if it finds any
+	 * rows under current snapshot that wouldn't be visible per the
+	 * transaction snapshot).
+	 */
+	if (XactIsoLevel == XACT_SERIALIZABLE)
+	{
+		useCurrentSnapshot = detectNewRows;
+	}
+	else
+	{
+		SetQuerySnapshot();
+		useCurrentSnapshot = false;
+	}
 
 	/*
 	 * If this is a select query (e.g., for a 'no action' or 'restrict'
@@ -2854,19 +2886,20 @@ ri_PerformCheck(RI_QueryKey *qkey, void *qplan,
 	 */
 	limit = (expect_OK == SPI_OK_SELECT) ? 1 : 0;
 
-	/*
-	 * Run the plan, using SnapshotNow time qual rules so that we can see
-	 * all committed tuples, even those committed after our own transaction
-	 * or query started.
-	 */
-	spi_result = SPI_execp_now(qplan, vals, nulls, limit);
+	/* Switch to proper UID to perform check as */
+	save_uid = GetUserId();
+	SetUserId(RelationGetForm(query_rel)->relowner);
+
+	/* Finally we can run the query. */
+	spi_result = SPI_execp_current(qplan, vals, nulls,
+								   useCurrentSnapshot, limit);
 
 	/* Restore UID */
 	SetUserId(save_uid);
 
 	/* Check result */
 	if (spi_result < 0)
-		elog(ERROR, "SPI_execp_now returned %d", spi_result);
+		elog(ERROR, "SPI_execp_current returned %d", spi_result);
 
 	if (expect_OK >= 0 && spi_result != expect_OK)
 		ri_ReportViolation(qkey, constrname ? constrname : "",

src/backend/utils/time/tqual.c

@@ -16,7 +16,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/utils/time/tqual.c,v 1.69 2003/09/25 18:58:35 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/utils/time/tqual.c,v 1.70 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -26,14 +26,19 @@
 #include "storage/sinval.h"
 #include "utils/tqual.h"
 
-
-static SnapshotData SnapshotDirtyData;
-Snapshot	SnapshotDirty = &SnapshotDirtyData;
-
+/*
+ * The SnapshotData structs are static to simplify memory allocation
+ * (see the hack in GetSnapshotData to avoid repeated malloc/free).
+ */
 static SnapshotData QuerySnapshotData;
 static SnapshotData SerializableSnapshotData;
+static SnapshotData CurrentSnapshotData;
+static SnapshotData SnapshotDirtyData;
+
+/* Externally visible pointers to valid snapshots: */
 Snapshot	QuerySnapshot = NULL;
 Snapshot	SerializableSnapshot = NULL;
+Snapshot	SnapshotDirty = &SnapshotDirtyData;
 
 /* These are updated by GetSnapshotData: */
 TransactionId RecentXmin = InvalidTransactionId;
@@ -387,10 +392,8 @@ HeapTupleSatisfiesToast(HeapTupleHeader tuple)
  *	CurrentCommandId.
  */
 int
-HeapTupleSatisfiesUpdate(HeapTuple htuple, CommandId curcid)
+HeapTupleSatisfiesUpdate(HeapTupleHeader tuple, CommandId curcid)
 {
-	HeapTupleHeader tuple = htuple->t_data;
-
 	if (!(tuple->t_infomask & HEAP_XMIN_COMMITTED))
 	{
 		if (tuple->t_infomask & HEAP_XMIN_INVALID)
@@ -1023,6 +1026,42 @@ CopyQuerySnapshot(void)
 	return snapshot;
 }
 
+/*
+ * CopyCurrentSnapshot
+ *		Make a snapshot that is up-to-date as of the current instant,
+ *		and return a copy.
+ *
+ * The copy is palloc'd in the current memory context.
+ */
+Snapshot
+CopyCurrentSnapshot(void)
+{
+	Snapshot	currentSnapshot;
+	Snapshot	snapshot;
+
+	if (QuerySnapshot == NULL)	/* should not be first call in xact */
+		elog(ERROR, "no snapshot has been set");
+
+	/* Update the static struct */
+	currentSnapshot = GetSnapshotData(&CurrentSnapshotData, false);
+	currentSnapshot->curcid = GetCurrentCommandId();
+
+	/* Make a copy */
+	snapshot = (Snapshot) palloc(sizeof(SnapshotData));
+	memcpy(snapshot, currentSnapshot, sizeof(SnapshotData));
+	if (snapshot->xcnt > 0)
+	{
+		snapshot->xip = (TransactionId *)
+			palloc(snapshot->xcnt * sizeof(TransactionId));
+		memcpy(snapshot->xip, currentSnapshot->xip,
+			   snapshot->xcnt * sizeof(TransactionId));
+	}
+	else
+		snapshot->xip = NULL;
+
+	return snapshot;
+}
+
 /*
  * FreeXactSnapshot
  *		Free snapshot(s) at end of transaction.
@@ -1031,8 +1070,9 @@ void
 FreeXactSnapshot(void)
 {
 	/*
-	 * We do not free(QuerySnapshot->xip); or
-	 * free(SerializableSnapshot->xip); they will be reused soon
+	 * We do not free the xip arrays for the snapshot structs;
+	 * they will be reused soon.  So this is now just a state
+	 * change to prevent outside callers from accessing the snapshots.
 	 */
 	QuerySnapshot = NULL;
 	SerializableSnapshot = NULL;

src/include/access/heapam.h

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Id: heapam.h,v 1.84 2003/09/15 23:33:43 tgl Exp $
+ * $Id: heapam.h,v 1.85 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -155,9 +155,9 @@ extern void setLastTid(const ItemPointer tid);
 
 extern Oid	heap_insert(Relation relation, HeapTuple tup, CommandId cid);
 extern int heap_delete(Relation relation, ItemPointer tid, ItemPointer ctid,
-			CommandId cid, bool wait);
+			CommandId cid, Snapshot crosscheck, bool wait);
 extern int heap_update(Relation relation, ItemPointer otid, HeapTuple tup,
-			ItemPointer ctid, CommandId cid, bool wait);
+			ItemPointer ctid, CommandId cid, Snapshot crosscheck, bool wait);
 extern int heap_mark4update(Relation relation, HeapTuple tup,
 				 Buffer *userbuf, CommandId cid);
 

src/include/executor/executor.h

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Id: executor.h,v 1.101 2003/09/25 18:58:35 tgl Exp $
+ * $Id: executor.h,v 1.102 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -85,7 +85,7 @@ extern HeapTuple ExecRemoveJunk(JunkFilter *junkfilter, TupleTableSlot *slot);
 /*
  * prototypes from functions in execMain.c
  */
-extern void ExecutorStart(QueryDesc *queryDesc, bool useSnapshotNow,
+extern void ExecutorStart(QueryDesc *queryDesc, bool useCurrentSnapshot,
 						  bool explainOnly);
 extern TupleTableSlot *ExecutorRun(QueryDesc *queryDesc,
 			ScanDirection direction, long count);

src/include/executor/spi.h

@@ -2,7 +2,7 @@
  *
  * spi.h
  *
- * $Id: spi.h,v 1.38 2003/09/25 18:58:36 tgl Exp $
+ * $Id: spi.h,v 1.39 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -84,8 +84,8 @@ extern void SPI_pop(void);
 extern int	SPI_exec(const char *src, int tcount);
 extern int SPI_execp(void *plan, Datum *values, const char *Nulls,
 		  int tcount);
-extern int SPI_execp_now(void *plan, Datum *values, const char *Nulls,
-		  int tcount);
+extern int SPI_execp_current(void *plan, Datum *values, const char *Nulls,
+							 bool useCurrentSnapshot, int tcount);
 extern void *SPI_prepare(const char *src, int nargs, Oid *argtypes);
 extern void *SPI_saveplan(void *plan);
 extern int	SPI_freeplan(void *plan);

src/include/nodes/execnodes.h

@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Id: execnodes.h,v 1.106 2003/09/25 18:58:36 tgl Exp $
+ * $Id: execnodes.h,v 1.107 2003/10/01 21:30:52 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -286,7 +286,7 @@ typedef struct EState
 	/* Basic state for all query types: */
 	ScanDirection es_direction; /* current scan direction */
 	Snapshot	es_snapshot;	/* time qual to use */
-	CommandId	es_snapshot_cid;	/* CommandId component of time qual */
+	Snapshot	es_crosscheck_snapshot;	/* crosscheck time qual for RI */
 	List	   *es_range_table; /* List of RangeTableEntrys */
 
 	/* Info about target table for insert/update/delete queries: */

src/include/utils/tqual.h

@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1996-2003, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Id: tqual.h,v 1.47 2003/09/25 18:58:36 tgl Exp $
+ * $Id: tqual.h,v 1.48 2003/10/01 21:30:53 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -100,7 +100,7 @@ extern bool HeapTupleSatisfiesDirty(HeapTupleHeader tuple);
 extern bool HeapTupleSatisfiesToast(HeapTupleHeader tuple);
 extern bool HeapTupleSatisfiesSnapshot(HeapTupleHeader tuple,
 						   Snapshot snapshot);
-extern int HeapTupleSatisfiesUpdate(HeapTuple tuple,
+extern int HeapTupleSatisfiesUpdate(HeapTupleHeader tuple,
 						 CommandId curcid);
 extern HTSV_Result HeapTupleSatisfiesVacuum(HeapTupleHeader tuple,
 						 TransactionId OldestXmin);
@@ -108,6 +108,7 @@ extern HTSV_Result HeapTupleSatisfiesVacuum(HeapTupleHeader tuple,
 extern Snapshot GetSnapshotData(Snapshot snapshot, bool serializable);
 extern void SetQuerySnapshot(void);
 extern Snapshot CopyQuerySnapshot(void);
+extern Snapshot CopyCurrentSnapshot(void);
 extern void FreeXactSnapshot(void);
 
 #endif   /* TQUAL_H */