Evade extra table_tuple_fetch_row_version() in ExecUpdate()/ExecDelete()

When we lock tuple using table_tuple_lock() then we at the same time fetch
the locked tuple to the slot.  In this case we can skip extra
table_tuple_fetch_row_version() thank to we've already fetched the 'old' tuple
and nobody can change it concurrently since it's locked.

Discussion: https://postgr.es/m/CAPpHfdua-YFw3XTprfutzGp28xXLigFtzNbuFY8yPhqeq6X5kg%40mail.gmail.com
Reviewed-by: Aleksander Alekseev, Pavel Borisov, Vignesh C, Mason Sharp
Reviewed-by: Andres Freund, Chris Travers

Author: akorotkov

src/backend/executor/nodeModifyTable.c

@@ -1559,6 +1559,22 @@ ExecDelete(ModifyTableContext *context,
 					{
 						case TM_Ok:
 							Assert(context->tmfd.traversed);
+
+							/*
+							 * Save locked tuple for further processing of
+							 * RETURNING clause.
+							 */
+							if (processReturning &&
+								resultRelInfo->ri_projectReturning &&
+								!resultRelInfo->ri_FdwRoutine)
+							{
+								TupleTableSlot *returningSlot;
+
+								returningSlot = ExecGetReturningSlot(estate, resultRelInfo);
+								ExecCopySlot(returningSlot, inputslot);
+								ExecMaterializeSlot(returningSlot);
+							}
+
 							epqslot = EvalPlanQual(context->epqstate,
 												   resultRelationDesc,
 												   resultRelInfo->ri_RangeTableIndex,
@@ -1673,12 +1689,17 @@ ExecDelete(ModifyTableContext *context,
 		}
 		else
 		{
+			/*
+			 * Tuple can be already fetched to the returning slot in case
+			 * we've previously locked it.  Fetch the tuple only if the slot
+			 * is empty.
+			 */
 			slot = ExecGetReturningSlot(estate, resultRelInfo);
 			if (oldtuple != NULL)
 			{
 				ExecForceStoreHeapTuple(oldtuple, slot, false);
 			}
-			else
+			else if (TupIsNull(slot))
 			{
 				if (!table_tuple_fetch_row_version(resultRelationDesc, tupleid,
 												   SnapshotAny, slot))
@@ -2393,6 +2414,19 @@ ExecUpdate(ModifyTableContext *context, ResultRelInfo *resultRelInfo,
 						case TM_Ok:
 							Assert(context->tmfd.traversed);
 
+							/* Make sure ri_oldTupleSlot is initialized. */
+							if (unlikely(!resultRelInfo->ri_projectNewInfoValid))
+								ExecInitUpdateProjection(context->mtstate,
+														 resultRelInfo);
+
+							/*
+							 * Save the locked tuple for further calculation
+							 * of the new tuple.
+							 */
+							oldSlot = resultRelInfo->ri_oldTupleSlot;
+							ExecCopySlot(oldSlot, inputslot);
+							ExecMaterializeSlot(oldSlot);
+
 							epqslot = EvalPlanQual(context->epqstate,
 												   resultRelationDesc,
 												   resultRelInfo->ri_RangeTableIndex,
@@ -2401,18 +2435,6 @@ ExecUpdate(ModifyTableContext *context, ResultRelInfo *resultRelInfo,
 								/* Tuple not passing quals anymore, exiting... */
 								return NULL;
 
-							/* Make sure ri_oldTupleSlot is initialized. */
-							if (unlikely(!resultRelInfo->ri_projectNewInfoValid))
-								ExecInitUpdateProjection(context->mtstate,
-														 resultRelInfo);
-
-							/* Fetch the most recent version of old tuple. */
-							oldSlot = resultRelInfo->ri_oldTupleSlot;
-							if (!table_tuple_fetch_row_version(resultRelationDesc,
-															   tupleid,
-															   SnapshotAny,
-															   oldSlot))
-								elog(ERROR, "failed to fetch tuple being updated");
 							slot = ExecGetUpdateNewTuple(resultRelInfo,
 														 epqslot, oldSlot);
 							goto redo_act;