v8-0003-Generalize-table-AM-API-for-INSERT-.-ON-CONFLICT.patch

From 8ad65fded0b1b7a9825920f4f17293732bf1b436 Mon Sep 17 00:00:00 2001
From: Alexander Korotkov <akorotkov@postgresql.org>
Date: Fri, 9 Jun 2023 00:05:52 +0300
Subject: [PATCH v8 3/8] Generalize table AM API for INSERT ... ON CONFLICT ...

Currently, all table AMs need to implement INSERT ... ON CONFLICT ... with
speculative tokens.  They could just have a custom implementation of those
tokens using tuple_insert_speculative() and tuple_complete_speculative() API
functions.

This commit changes INSERT ... ON CONFLICT ... implementation to use single
tuple_insert_with_arbiter() API function, which encapsulates the whole
alogrithm.  This new function provides clear semantics to make different
implementations of INSERT ... ON CONFLICT ... functionality.
---
 src/backend/access/heap/heapam_handler.c | 281 ++++++++++++++++++++++-
 src/backend/access/table/tableamapi.c    |   3 +-
 src/backend/executor/nodeModifyTable.c   | 270 ++--------------------
 src/include/access/tableam.h             |  84 +++----
 4 files changed, 348 insertions(+), 290 deletions(-)

diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index 26b3be9779..590413bab9 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -304,6 +304,284 @@ heapam_tuple_complete_speculative(Relation relation, TupleTableSlot *slot,
 		pfree(tuple);
 }
 
+/*
+ * ExecCheckTupleVisible -- verify tuple is visible
+ *
+ * It would not be consistent with guarantees of the higher isolation levels to
+ * proceed with avoiding insertion (taking speculative insertion's alternative
+ * path) on the basis of another tuple that is not visible to MVCC snapshot.
+ * Check for the need to raise a serialization failure, and do so as necessary.
+ */
+static void
+ExecCheckTupleVisible(EState *estate,
+					  Relation rel,
+					  TupleTableSlot *slot)
+{
+	if (!IsolationUsesXactSnapshot())
+		return;
+
+	if (!table_tuple_satisfies_snapshot(rel, slot, estate->es_snapshot))
+	{
+		Datum		xminDatum;
+		TransactionId xmin;
+		bool		isnull;
+
+		xminDatum = slot_getsysattr(slot, MinTransactionIdAttributeNumber, &isnull);
+		Assert(!isnull);
+		xmin = DatumGetTransactionId(xminDatum);
+
+		/*
+		 * We should not raise a serialization failure if the conflict is
+		 * against a tuple inserted by our own transaction, even if it's not
+		 * visible to our snapshot.  (This would happen, for example, if
+		 * conflicting keys are proposed for insertion in a single command.)
+		 */
+		if (!TransactionIdIsCurrentTransactionId(xmin))
+			ereport(ERROR,
+					(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+					 errmsg("could not serialize access due to concurrent update")));
+	}
+}
+
+/*
+ * ExecCheckTIDVisible -- convenience variant of ExecCheckTupleVisible()
+ */
+static void
+ExecCheckTIDVisible(EState *estate,
+					Relation rel,
+					ItemPointer tid,
+					TupleTableSlot *tempSlot)
+{
+	/* Redundantly check isolation level */
+	if (!IsolationUsesXactSnapshot())
+		return;
+
+	if (!table_tuple_fetch_row_version(rel, tid,
+									   SnapshotAny, tempSlot))
+		elog(ERROR, "failed to fetch conflicting tuple for ON CONFLICT");
+	ExecCheckTupleVisible(estate, rel, tempSlot);
+	ExecClearTuple(tempSlot);
+}
+
+static inline TupleTableSlot *
+heapam_tuple_insert_with_arbiter(ResultRelInfo *resultRelInfo,
+								 TupleTableSlot *slot,
+								 CommandId cid, int options,
+								 struct BulkInsertStateData *bistate,
+								 List *arbiterIndexes,
+								 EState *estate,
+								 LockTupleMode lockmode,
+								 TupleTableSlot *lockedSlot,
+								 TupleTableSlot *tempSlot)
+{
+	Relation	rel = resultRelInfo->ri_RelationDesc;
+	uint32		specToken;
+	ItemPointerData conflictTid;
+	bool		specConflict;
+	List	   *recheckIndexes = NIL;
+
+	while (true)
+	{
+		specConflict = false;
+		if (!ExecCheckIndexConstraints(resultRelInfo, slot, estate, &conflictTid,
+									   arbiterIndexes))
+		{
+			if (lockedSlot)
+			{
+				TM_Result	test;
+				TM_FailureData tmfd;
+				Datum		xminDatum;
+				TransactionId xmin;
+				bool		isnull;
+
+				/* Determine lock mode to use */
+				lockmode = ExecUpdateLockMode(estate, resultRelInfo);
+
+				/*
+				 * Lock tuple for update.  Don't follow updates when tuple
+				 * cannot be locked without doing so.  A row locking conflict
+				 * here means our previous conclusion that the tuple is
+				 * conclusively committed is not true anymore.
+				 */
+				test = table_tuple_lock(rel, &conflictTid,
+										estate->es_snapshot,
+										lockedSlot, estate->es_output_cid,
+										lockmode, LockWaitBlock, 0,
+										&tmfd);
+				switch (test)
+				{
+					case TM_Ok:
+						/* success! */
+						break;
+
+					case TM_Invisible:
+
+						/*
+						 * This can occur when a just inserted tuple is
+						 * updated again in the same command. E.g. because
+						 * multiple rows with the same conflicting key values
+						 * are inserted.
+						 *
+						 * This is somewhat similar to the ExecUpdate()
+						 * TM_SelfModified case.  We do not want to proceed
+						 * because it would lead to the same row being updated
+						 * a second time in some unspecified order, and in
+						 * contrast to plain UPDATEs there's no historical
+						 * behavior to break.
+						 *
+						 * It is the user's responsibility to prevent this
+						 * situation from occurring.  These problems are why
+						 * the SQL standard similarly specifies that for SQL
+						 * MERGE, an exception must be raised in the event of
+						 * an attempt to update the same row twice.
+						 */
+						xminDatum = slot_getsysattr(lockedSlot,
+													MinTransactionIdAttributeNumber,
+													&isnull);
+						Assert(!isnull);
+						xmin = DatumGetTransactionId(xminDatum);
+
+						if (TransactionIdIsCurrentTransactionId(xmin))
+							ereport(ERROR,
+									(errcode(ERRCODE_CARDINALITY_VIOLATION),
+							/* translator: %s is a SQL command name */
+									 errmsg("%s command cannot affect row a second time",
+											"ON CONFLICT DO UPDATE"),
+									 errhint("Ensure that no rows proposed for insertion within the same command have duplicate constrained values.")));
+
+						/* This shouldn't happen */
+						elog(ERROR, "attempted to lock invisible tuple");
+						break;
+
+					case TM_SelfModified:
+
+						/*
+						 * This state should never be reached. As a dirty
+						 * snapshot is used to find conflicting tuples,
+						 * speculative insertion wouldn't have seen this row
+						 * to conflict with.
+						 */
+						elog(ERROR, "unexpected self-updated tuple");
+						break;
+
+					case TM_Updated:
+						if (IsolationUsesXactSnapshot())
+							ereport(ERROR,
+									(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+									 errmsg("could not serialize access due to concurrent update")));
+
+						/*
+						 * As long as we don't support an UPDATE of INSERT ON
+						 * CONFLICT for a partitioned table we shouldn't reach
+						 * to a case where tuple to be lock is moved to
+						 * another partition due to concurrent update of the
+						 * partition key.
+						 */
+						Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
+
+						/*
+						 * Tell caller to try again from the very start.
+						 *
+						 * It does not make sense to use the usual
+						 * EvalPlanQual() style loop here, as the new version
+						 * of the row might not conflict anymore, or the
+						 * conflicting tuple has actually been deleted.
+						 */
+						ExecClearTuple(lockedSlot);
+						return false;
+
+					case TM_Deleted:
+						if (IsolationUsesXactSnapshot())
+							ereport(ERROR,
+									(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+									 errmsg("could not serialize access due to concurrent delete")));
+
+						/* see TM_Updated case */
+						Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
+						ExecClearTuple(lockedSlot);
+						return false;
+
+					default:
+						elog(ERROR, "unrecognized table_tuple_lock status: %u", test);
+				}
+
+				/* Success, the tuple is locked. */
+
+				/*
+				 * Verify that the tuple is visible to our MVCC snapshot if
+				 * the current isolation level mandates that.
+				 *
+				 * It's not sufficient to rely on the check within
+				 * ExecUpdate() as e.g. CONFLICT ... WHERE clause may prevent
+				 * us from reaching that.
+				 *
+				 * This means we only ever continue when a new command in the
+				 * current transaction could see the row, even though in READ
+				 * COMMITTED mode the tuple will not be visible according to
+				 * the current statement's snapshot.  This is in line with the
+				 * way UPDATE deals with newer tuple versions.
+				 */
+				ExecCheckTupleVisible(estate, rel, lockedSlot);
+				return NULL;
+			}
+			else
+			{
+				ExecCheckTIDVisible(estate, rel, &conflictTid, tempSlot);
+				return NULL;
+			}
+		}
+
+		/*
+		 * Before we start insertion proper, acquire our "speculative
+		 * insertion lock".  Others can use that to wait for us to decide if
+		 * we're going to go ahead with the insertion, instead of waiting for
+		 * the whole transaction to complete.
+		 */
+		specToken = SpeculativeInsertionLockAcquire(GetCurrentTransactionId());
+
+		/* insert the tuple, with the speculative token */
+		heapam_tuple_insert_speculative(rel, slot,
+										estate->es_output_cid,
+										0,
+										NULL,
+										specToken);
+
+		/* insert index entries for tuple */
+		recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
+											   slot, estate, false, true,
+											   &specConflict,
+											   arbiterIndexes,
+											   false);
+
+		/* adjust the tuple's state accordingly */
+		heapam_tuple_complete_speculative(rel, slot,
+										  specToken, !specConflict);
+
+		/*
+		 * Wake up anyone waiting for our decision.  They will re-check the
+		 * tuple, see that it's no longer speculative, and wait on our XID as
+		 * if this was a regularly inserted tuple all along.  Or if we killed
+		 * the tuple, they will see it's dead, and proceed as if the tuple
+		 * never existed.
+		 */
+		SpeculativeInsertionLockRelease(GetCurrentTransactionId());
+
+		/*
+		 * If there was a conflict, start from the beginning.  We'll do the
+		 * pre-check again, which will now find the conflicting tuple (unless
+		 * it aborts before we get there).
+		 */
+		if (specConflict)
+		{
+			list_free(recheckIndexes);
+			CHECK_FOR_INTERRUPTS();
+			continue;
+		}
+
+		return slot;
+	}
+}
+
 static TM_Result
 heapam_tuple_delete(Relation relation, ItemPointer tid, CommandId cid,
 					Snapshot snapshot, Snapshot crosscheck, int options,
@@ -2644,8 +2922,7 @@ static const TableAmRoutine heapam_methods = {
 	.index_fetch_tuple = heapam_index_fetch_tuple,
 
 	.tuple_insert = heapam_tuple_insert,
-	.tuple_insert_speculative = heapam_tuple_insert_speculative,
-	.tuple_complete_speculative = heapam_tuple_complete_speculative,
+	.tuple_insert_with_arbiter = heapam_tuple_insert_with_arbiter,
 	.multi_insert = heap_multi_insert,
 	.tuple_delete = heapam_tuple_delete,
 	.tuple_update = heapam_tuple_update,
diff --git a/src/backend/access/table/tableamapi.c b/src/backend/access/table/tableamapi.c
index d9e23ef317..c38ab936cd 100644
--- a/src/backend/access/table/tableamapi.c
+++ b/src/backend/access/table/tableamapi.c
@@ -70,8 +70,7 @@ GetTableAmRoutine(Oid amhandler)
 	 * Could be made optional, but would require throwing error during
 	 * parse-analysis.
 	 */
-	Assert(routine->tuple_insert_speculative != NULL);
-	Assert(routine->tuple_complete_speculative != NULL);
+	Assert(routine->tuple_insert_with_arbiter != NULL);
 
 	Assert(routine->multi_insert != NULL);
 	Assert(routine->tuple_delete != NULL);
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index d1917f2fea..8e1c8f697c 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -129,7 +129,6 @@ static void ExecCrossPartitionUpdateForeignKey(ModifyTableContext *context,
 											   TupleTableSlot *newslot);
 static bool ExecOnConflictUpdate(ModifyTableContext *context,
 								 ResultRelInfo *resultRelInfo,
-								 ItemPointer conflictTid,
 								 TupleTableSlot *excludedSlot,
 								 bool canSetTag,
 								 TupleTableSlot **returning);
@@ -265,66 +264,6 @@ ExecProcessReturning(ResultRelInfo *resultRelInfo,
 	return ExecProject(projectReturning);
 }
 
-/*
- * ExecCheckTupleVisible -- verify tuple is visible
- *
- * It would not be consistent with guarantees of the higher isolation levels to
- * proceed with avoiding insertion (taking speculative insertion's alternative
- * path) on the basis of another tuple that is not visible to MVCC snapshot.
- * Check for the need to raise a serialization failure, and do so as necessary.
- */
-static void
-ExecCheckTupleVisible(EState *estate,
-					  Relation rel,
-					  TupleTableSlot *slot)
-{
-	if (!IsolationUsesXactSnapshot())
-		return;
-
-	if (!table_tuple_satisfies_snapshot(rel, slot, estate->es_snapshot))
-	{
-		Datum		xminDatum;
-		TransactionId xmin;
-		bool		isnull;
-
-		xminDatum = slot_getsysattr(slot, MinTransactionIdAttributeNumber, &isnull);
-		Assert(!isnull);
-		xmin = DatumGetTransactionId(xminDatum);
-
-		/*
-		 * We should not raise a serialization failure if the conflict is
-		 * against a tuple inserted by our own transaction, even if it's not
-		 * visible to our snapshot.  (This would happen, for example, if
-		 * conflicting keys are proposed for insertion in a single command.)
-		 */
-		if (!TransactionIdIsCurrentTransactionId(xmin))
-			ereport(ERROR,
-					(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
-					 errmsg("could not serialize access due to concurrent update")));
-	}
-}
-
-/*
- * ExecCheckTIDVisible -- convenience variant of ExecCheckTupleVisible()
- */
-static void
-ExecCheckTIDVisible(EState *estate,
-					ResultRelInfo *relinfo,
-					ItemPointer tid,
-					TupleTableSlot *tempSlot)
-{
-	Relation	rel = relinfo->ri_RelationDesc;
-
-	/* Redundantly check isolation level */
-	if (!IsolationUsesXactSnapshot())
-		return;
-
-	if (!table_tuple_fetch_row_version(rel, tid, SnapshotAny, tempSlot))
-		elog(ERROR, "failed to fetch conflicting tuple for ON CONFLICT");
-	ExecCheckTupleVisible(estate, rel, tempSlot);
-	ExecClearTuple(tempSlot);
-}
-
 /*
  * Initialize to compute stored generated columns for a tuple
  *
@@ -1015,12 +954,19 @@ ExecInsert(ModifyTableContext *context,
 		if (onconflict != ONCONFLICT_NONE && resultRelInfo->ri_NumIndices > 0)
 		{
 			/* Perform a speculative insertion. */
-			uint32		specToken;
-			ItemPointerData conflictTid;
-			bool		specConflict;
 			List	   *arbiterIndexes;
+			TupleTableSlot *existing = NULL,
+					   *returningSlot,
+					   *inserted;
+			LockTupleMode lockmode = LockTupleExclusive;
 
 			arbiterIndexes = resultRelInfo->ri_onConflictArbiterIndexes;
+			returningSlot = ExecGetReturningSlot(estate, resultRelInfo);
+			if (onconflict == ONCONFLICT_UPDATE)
+			{
+				lockmode = ExecUpdateLockMode(estate, resultRelInfo);
+				existing = resultRelInfo->ri_onConflict->oc_Existing;
+			}
 
 			/*
 			 * Do a non-conclusive check for conflicts first.
@@ -1037,23 +983,28 @@ ExecInsert(ModifyTableContext *context,
 			 */
 	vlock:
 			CHECK_FOR_INTERRUPTS();
-			specConflict = false;
-			if (!ExecCheckIndexConstraints(resultRelInfo, slot, estate,
-										   &conflictTid, arbiterIndexes))
+			inserted = table_tuple_insert_with_arbiter(resultRelInfo,
+													   slot, estate->es_output_cid,
+													   0, NULL, arbiterIndexes, estate,
+													   lockmode, existing, returningSlot);
+			if (!inserted)
 			{
 				/* committed conflict tuple found */
 				if (onconflict == ONCONFLICT_UPDATE)
 				{
+					TupleTableSlot *returning = NULL;
+
+					if (TTS_EMPTY(existing))
+						goto vlock;
+
 					/*
 					 * In case of ON CONFLICT DO UPDATE, execute the UPDATE
 					 * part.  Be prepared to retry if the UPDATE fails because
 					 * of another concurrent UPDATE/DELETE to the conflict
 					 * tuple.
 					 */
-					TupleTableSlot *returning = NULL;
-
 					if (ExecOnConflictUpdate(context, resultRelInfo,
-											 &conflictTid, slot, canSetTag,
+											 slot, canSetTag,
 											 &returning))
 					{
 						InstrCountTuples2(&mtstate->ps, 1);
@@ -1076,57 +1027,13 @@ ExecInsert(ModifyTableContext *context,
 					 * ExecGetReturningSlot() in the DO NOTHING case...
 					 */
 					Assert(onconflict == ONCONFLICT_NOTHING);
-					ExecCheckTIDVisible(estate, resultRelInfo, &conflictTid,
-										ExecGetReturningSlot(estate, resultRelInfo));
 					InstrCountTuples2(&mtstate->ps, 1);
 					return NULL;
 				}
 			}
-
-			/*
-			 * Before we start insertion proper, acquire our "speculative
-			 * insertion lock".  Others can use that to wait for us to decide
-			 * if we're going to go ahead with the insertion, instead of
-			 * waiting for the whole transaction to complete.
-			 */
-			specToken = SpeculativeInsertionLockAcquire(GetCurrentTransactionId());
-
-			/* insert the tuple, with the speculative token */
-			table_tuple_insert_speculative(resultRelationDesc, slot,
-										   estate->es_output_cid,
-										   0,
-										   NULL,
-										   specToken);
-
-			/* insert index entries for tuple */
-			recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
-												   slot, estate, false, true,
-												   &specConflict,
-												   arbiterIndexes,
-												   false);
-
-			/* adjust the tuple's state accordingly */
-			table_tuple_complete_speculative(resultRelationDesc, slot,
-											 specToken, !specConflict);
-
-			/*
-			 * Wake up anyone waiting for our decision.  They will re-check
-			 * the tuple, see that it's no longer speculative, and wait on our
-			 * XID as if this was a regularly inserted tuple all along.  Or if
-			 * we killed the tuple, they will see it's dead, and proceed as if
-			 * the tuple never existed.
-			 */
-			SpeculativeInsertionLockRelease(GetCurrentTransactionId());
-
-			/*
-			 * If there was a conflict, start from the beginning.  We'll do
-			 * the pre-check again, which will now find the conflicting tuple
-			 * (unless it aborts before we get there).
-			 */
-			if (specConflict)
+			else
 			{
-				list_free(recheckIndexes);
-				goto vlock;
+				slot = inserted;
 			}
 
 			/* Since there was no insertion conflict, we're done */
@@ -2441,144 +2348,15 @@ redo_act:
 static bool
 ExecOnConflictUpdate(ModifyTableContext *context,
 					 ResultRelInfo *resultRelInfo,
-					 ItemPointer conflictTid,
 					 TupleTableSlot *excludedSlot,
 					 bool canSetTag,
 					 TupleTableSlot **returning)
 {
 	ModifyTableState *mtstate = context->mtstate;
 	ExprContext *econtext = mtstate->ps.ps_ExprContext;
-	Relation	relation = resultRelInfo->ri_RelationDesc;
 	ExprState  *onConflictSetWhere = resultRelInfo->ri_onConflict->oc_WhereClause;
 	TupleTableSlot *existing = resultRelInfo->ri_onConflict->oc_Existing;
-	TM_FailureData tmfd;
-	LockTupleMode lockmode;
-	TM_Result	test;
-	Datum		xminDatum;
-	TransactionId xmin;
-	bool		isnull;
-
-	/* Determine lock mode to use */
-	lockmode = ExecUpdateLockMode(context->estate, resultRelInfo);
-
-	/*
-	 * Lock tuple for update.  Don't follow updates when tuple cannot be
-	 * locked without doing so.  A row locking conflict here means our
-	 * previous conclusion that the tuple is conclusively committed is not
-	 * true anymore.
-	 */
-	test = table_tuple_lock(relation, conflictTid,
-							context->estate->es_snapshot,
-							existing, context->estate->es_output_cid,
-							lockmode, LockWaitBlock, 0,
-							&tmfd);
-	switch (test)
-	{
-		case TM_Ok:
-			/* success! */
-			break;
-
-		case TM_Invisible:
-
-			/*
-			 * This can occur when a just inserted tuple is updated again in
-			 * the same command. E.g. because multiple rows with the same
-			 * conflicting key values are inserted.
-			 *
-			 * This is somewhat similar to the ExecUpdate() TM_SelfModified
-			 * case.  We do not want to proceed because it would lead to the
-			 * same row being updated a second time in some unspecified order,
-			 * and in contrast to plain UPDATEs there's no historical behavior
-			 * to break.
-			 *
-			 * It is the user's responsibility to prevent this situation from
-			 * occurring.  These problems are why the SQL standard similarly
-			 * specifies that for SQL MERGE, an exception must be raised in
-			 * the event of an attempt to update the same row twice.
-			 */
-			xminDatum = slot_getsysattr(existing,
-										MinTransactionIdAttributeNumber,
-										&isnull);
-			Assert(!isnull);
-			xmin = DatumGetTransactionId(xminDatum);
-
-			if (TransactionIdIsCurrentTransactionId(xmin))
-				ereport(ERROR,
-						(errcode(ERRCODE_CARDINALITY_VIOLATION),
-				/* translator: %s is a SQL command name */
-						 errmsg("%s command cannot affect row a second time",
-								"ON CONFLICT DO UPDATE"),
-						 errhint("Ensure that no rows proposed for insertion within the same command have duplicate constrained values.")));
-
-			/* This shouldn't happen */
-			elog(ERROR, "attempted to lock invisible tuple");
-			break;
-
-		case TM_SelfModified:
-
-			/*
-			 * This state should never be reached. As a dirty snapshot is used
-			 * to find conflicting tuples, speculative insertion wouldn't have
-			 * seen this row to conflict with.
-			 */
-			elog(ERROR, "unexpected self-updated tuple");
-			break;
-
-		case TM_Updated:
-			if (IsolationUsesXactSnapshot())
-				ereport(ERROR,
-						(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
-						 errmsg("could not serialize access due to concurrent update")));
-
-			/*
-			 * As long as we don't support an UPDATE of INSERT ON CONFLICT for
-			 * a partitioned table we shouldn't reach to a case where tuple to
-			 * be lock is moved to another partition due to concurrent update
-			 * of the partition key.
-			 */
-			Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
-
-			/*
-			 * Tell caller to try again from the very start.
-			 *
-			 * It does not make sense to use the usual EvalPlanQual() style
-			 * loop here, as the new version of the row might not conflict
-			 * anymore, or the conflicting tuple has actually been deleted.
-			 */
-			ExecClearTuple(existing);
-			return false;
-
-		case TM_Deleted:
-			if (IsolationUsesXactSnapshot())
-				ereport(ERROR,
-						(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
-						 errmsg("could not serialize access due to concurrent delete")));
-
-			/* see TM_Updated case */
-			Assert(!ItemPointerIndicatesMovedPartitions(&tmfd.ctid));
-			ExecClearTuple(existing);
-			return false;
-
-		default:
-			elog(ERROR, "unrecognized table_tuple_lock status: %u", test);
-	}
-
-	/* Success, the tuple is locked. */
-
-	/*
-	 * Verify that the tuple is visible to our MVCC snapshot if the current
-	 * isolation level mandates that.
-	 *
-	 * It's not sufficient to rely on the check within ExecUpdate() as e.g.
-	 * CONFLICT ... WHERE clause may prevent us from reaching that.
-	 *
-	 * This means we only ever continue when a new command in the current
-	 * transaction could see the row, even though in READ COMMITTED mode the
-	 * tuple will not be visible according to the current statement's
-	 * snapshot.  This is in line with the way UPDATE deals with newer tuple
-	 * versions.
-	 */
-	ExecCheckTupleVisible(context->estate, relation, existing);
+	ItemPointer conflictTid = &existing->tts_tid;
 
 	/*
 	 * Make tuple and any needed join variables available to ExecQual and
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
index cf68ec48eb..c4cdae5903 100644
--- a/src/include/access/tableam.h
+++ b/src/include/access/tableam.h
@@ -22,6 +22,7 @@
 #include "access/xact.h"
 #include "commands/vacuum.h"
 #include "executor/tuptable.h"
+#include "nodes/execnodes.h"
 #include "utils/rel.h"
 #include "utils/snapshot.h"
 
@@ -514,19 +515,16 @@ typedef struct TableAmRoutine
 									 CommandId cid, int options,
 									 struct BulkInsertStateData *bistate);
 
-	/* see table_tuple_insert_speculative() for reference about parameters */
-	void		(*tuple_insert_speculative) (Relation rel,
-											 TupleTableSlot *slot,
-											 CommandId cid,
-											 int options,
-											 struct BulkInsertStateData *bistate,
-											 uint32 specToken);
-
-	/* see table_tuple_complete_speculative() for reference about parameters */
-	void		(*tuple_complete_speculative) (Relation rel,
-											   TupleTableSlot *slot,
-											   uint32 specToken,
-											   bool succeeded);
+	/* see table_tuple_insert_with_arbiter() for reference about parameters */
+	TupleTableSlot *(*tuple_insert_with_arbiter) (ResultRelInfo *resultRelInfo,
+												  TupleTableSlot *slot,
+												  CommandId cid, int options,
+												  struct BulkInsertStateData *bistate,
+												  List *arbiterIndexes,
+												  EState *estate,
+												  LockTupleMode lockmode,
+												  TupleTableSlot *lockedSlot,
+												  TupleTableSlot *tempSlot);
 
 	/* see table_multi_insert() for reference about parameters */
 	void		(*multi_insert) (Relation rel, TupleTableSlot **slots, int nslots,
@@ -1400,36 +1398,42 @@ table_tuple_insert(Relation rel, TupleTableSlot *slot, CommandId cid,
 }
 
 /*
- * Perform a "speculative insertion". These can be backed out afterwards
- * without aborting the whole transaction.  Other sessions can wait for the
- * speculative insertion to be confirmed, turning it into a regular tuple, or
- * aborted, as if it never existed.  Speculatively inserted tuples behave as
- * "value locks" of short duration, used to implement INSERT .. ON CONFLICT.
+ * Insert a tuple from a slot into table AM routine with arbiter indexes.
  *
- * A transaction having performed a speculative insertion has to either abort,
- * or finish the speculative insertion with
- * table_tuple_complete_speculative(succeeded = ...).
- */
-static inline void
-table_tuple_insert_speculative(Relation rel, TupleTableSlot *slot,
-							   CommandId cid, int options,
-							   struct BulkInsertStateData *bistate,
-							   uint32 specToken)
-{
-	rel->rd_tableam->tuple_insert_speculative(rel, slot, cid, options,
-											  bistate, specToken);
-}
-
-/*
- * Complete "speculative insertion" started in the same transaction. If
- * succeeded is true, the tuple is fully inserted, if false, it's removed.
+ * This function is similar to table_tuple_insert(), but it takes into account
+ * `arbiterIndexes`, which comprises the list of oids of arbiter indexes.
+ *
+ * If tuple doesn't violates uniqueness on all arbiter indexes, then it should
+ * be inserted and the slot containing inserted tuple is returned.
+ *
+ * If tuple violates uniqueness on any arbiter index, then this function
+ * returns NULL and doesn't insert the tuple.  Also, if 'lockedSlot' is
+ * provided, then conflicting tuple gets locked in `lockmode` and placed into
+ * `lockedSlot`.
+ *
+ * Executor state `estate` is passed to this method to provide ability to
+ * calculate index tuples.  Temporary tuple table slot `tempSlot` is passed
+ * for holding of potentially conflicing tuple.
  */
-static inline void
-table_tuple_complete_speculative(Relation rel, TupleTableSlot *slot,
-								 uint32 specToken, bool succeeded)
+static inline TupleTableSlot *
+table_tuple_insert_with_arbiter(ResultRelInfo *resultRelInfo,
+								TupleTableSlot *slot,
+								CommandId cid, int options,
+								struct BulkInsertStateData *bistate,
+								List *arbiterIndexes,
+								EState *estate,
+								LockTupleMode lockmode,
+								TupleTableSlot *lockedSlot,
+								TupleTableSlot *tempSlot)
 {
-	rel->rd_tableam->tuple_complete_speculative(rel, slot, specToken,
-												succeeded);
+	Relation	rel = resultRelInfo->ri_RelationDesc;
+
+	return rel->rd_tableam->tuple_insert_with_arbiter(resultRelInfo,
+													  slot, cid, options,
+													  bistate, arbiterIndexes,
+													  estate,
+													  lockmode, lockedSlot,
+													  tempSlot);
 }
 
 /*
-- 
2.39.2 (Apple Git-143)