Use slots more widely in tuple mapping code and make naming more consistent.

It's inefficient to use a single slot for mapping between tuple
descriptors for multiple tuples, as previously done when using
ConvertPartitionTupleSlot(), as that means the slot's tuple descriptors
change for every tuple.

Previously we also, via ConvertPartitionTupleSlot(), built new tuples
after the mapping even in cases where we, immediately afterwards,
access individual columns again.

Refactor the code so one slot, on demand, is used for each
partition. That avoids having to change the descriptor (and allows to
use the more efficient "fixed" tuple slots). Then use slot->slot
mapping, to avoid unnecessarily forming a tuple.

As the naming between the tuple and slot mapping functions wasn't
consistent, rename them to execute_attr_map_{tuple,slot}.  It's likely
that we'll also rename convert_tuples_by_* to denote that these
functions "only" build a map, but that's left for later.

Author: Amit Khandekar and Amit Langote, editorialized by me
Reviewed-By: Amit Langote, Amit Khandekar, Andres Freund
Discussion:
    https://postgr.es/m/CAJ3gD9fR0wRNeAE8VqffNTyONS_UfFPRpqxhnD9Q42vZB+Jvpg@mail.gmail.com
    https://postgr.es/m/e4f9d743-cd4b-efb0-7574-da21d86a7f36%40lab.ntt.co.jp
Backpatch: -

src/backend/access/common/tupconvert.c

@@ -4,10 +4,8 @@
  *	  Tuple conversion support.
  *
  * These functions provide conversion between rowtypes that are logically
- * equivalent but might have columns in a different order or different sets
- * of dropped columns.  There is some overlap of functionality with the
- * executor's "junkfilter" routines, but these functions work on bare
- * HeapTuples rather than TupleTableSlots.
+ * equivalent but might have columns in a different order or different sets of
+ * dropped columns.
  *
  * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
@@ -22,6 +20,7 @@
 
 #include "access/htup_details.h"
 #include "access/tupconvert.h"
+#include "executor/tuptable.h"
 #include "utils/builtins.h"
 
 
@@ -31,7 +30,7 @@
  * The setup routine checks whether the given source and destination tuple
  * descriptors are logically compatible.  If not, it throws an error.
  * If so, it returns NULL if they are physically compatible (ie, no conversion
- * is needed), else a TupleConversionMap that can be used by do_convert_tuple
+ * is needed), else a TupleConversionMap that can be used by execute_attr_map_tuple
  * to perform the conversion.
  *
  * The TupleConversionMap, if needed, is palloc'd in the caller's memory
@@ -214,55 +213,13 @@ convert_tuples_by_name(TupleDesc indesc,
 	TupleConversionMap *map;
 	AttrNumber *attrMap;
 	int			n = outdesc->natts;
-	int			i;
-	bool		same;
 
 	/* Verify compatibility and prepare attribute-number map */
-	attrMap = convert_tuples_by_name_map(indesc, outdesc, msg);
-
-	/*
-	 * Check to see if the map is one-to-one, in which case we need not do a
-	 * tuple conversion.  We must also insist that both tupdescs either
-	 * specify or don't specify an OID column, else we need a conversion to
-	 * add/remove space for that.  (For some callers, presence or absence of
-	 * an OID column perhaps would not really matter, but let's be safe.)
-	 */
-	if (indesc->natts == outdesc->natts &&
-		indesc->tdhasoid == outdesc->tdhasoid)
-	{
-		same = true;
-		for (i = 0; i < n; i++)
-		{
-			Form_pg_attribute inatt;
-			Form_pg_attribute outatt;
-
-			if (attrMap[i] == (i + 1))
-				continue;
-
-			/*
-			 * If it's a dropped column and the corresponding input column is
-			 * also dropped, we needn't convert.  However, attlen and attalign
-			 * must agree.
-			 */
-			inatt = TupleDescAttr(indesc, i);
-			outatt = TupleDescAttr(outdesc, i);
-			if (attrMap[i] == 0 &&
-				inatt->attisdropped &&
-				inatt->attlen == outatt->attlen &&
-				inatt->attalign == outatt->attalign)
-				continue;
-
-			same = false;
-			break;
-		}
-	}
-	else
-		same = false;
+	attrMap = convert_tuples_by_name_map_if_req(indesc, outdesc, msg);
 
-	if (same)
+	if (attrMap == NULL)
 	{
-		/* Runtime conversion is not needed */
-		pfree(attrMap);
+		/* runtime conversion is not needed */
 		return NULL;
 	}
 
@@ -367,11 +324,78 @@ convert_tuples_by_name_map(TupleDesc indesc,
 	return attrMap;
 }
 
+/*
+ * Returns mapping created by convert_tuples_by_name_map, or NULL if no
+ * conversion not required. This is a convenience routine for
+ * convert_tuples_by_name() and other functions.
+ */
+AttrNumber *
+convert_tuples_by_name_map_if_req(TupleDesc indesc,
+								  TupleDesc outdesc,
+								  const char *msg)
+{
+	AttrNumber *attrMap;
+	int			n = outdesc->natts;
+	int			i;
+	bool		same;
+
+	/* Verify compatibility and prepare attribute-number map */
+	attrMap = convert_tuples_by_name_map(indesc, outdesc, msg);
+
+	/*
+	 * Check to see if the map is one-to-one, in which case we need not do a
+	 * tuple conversion.  We must also insist that both tupdescs either
+	 * specify or don't specify an OID column, else we need a conversion to
+	 * add/remove space for that.  (For some callers, presence or absence of
+	 * an OID column perhaps would not really matter, but let's be safe.)
+	 */
+	if (indesc->natts == outdesc->natts &&
+		indesc->tdhasoid == outdesc->tdhasoid)
+	{
+		same = true;
+		for (i = 0; i < n; i++)
+		{
+			Form_pg_attribute inatt;
+			Form_pg_attribute outatt;
+
+			if (attrMap[i] == (i + 1))
+				continue;
+
+			/*
+			 * If it's a dropped column and the corresponding input column is
+			 * also dropped, we needn't convert.  However, attlen and attalign
+			 * must agree.
+			 */
+			inatt = TupleDescAttr(indesc, i);
+			outatt = TupleDescAttr(outdesc, i);
+			if (attrMap[i] == 0 &&
+				inatt->attisdropped &&
+				inatt->attlen == outatt->attlen &&
+				inatt->attalign == outatt->attalign)
+				continue;
+
+			same = false;
+			break;
+		}
+	}
+	else
+		same = false;
+
+	if (same)
+	{
+		/* Runtime conversion is not needed */
+		pfree(attrMap);
+		return NULL;
+	}
+	else
+		return attrMap;
+}
+
 /*
  * Perform conversion of a tuple according to the map.
  */
 HeapTuple
-do_convert_tuple(HeapTuple tuple, TupleConversionMap *map)
+execute_attr_map_tuple(HeapTuple tuple, TupleConversionMap *map)
 {
 	AttrNumber *attrMap = map->attrMap;
 	Datum	   *invalues = map->invalues;
@@ -405,6 +429,62 @@ do_convert_tuple(HeapTuple tuple, TupleConversionMap *map)
 	return heap_form_tuple(map->outdesc, outvalues, outisnull);
 }
 
+/*
+ * Perform conversion of a tuple slot according to the map.
+ */
+TupleTableSlot *
+execute_attr_map_slot(AttrNumber *attrMap,
+					  TupleTableSlot *in_slot,
+					  TupleTableSlot *out_slot)
+{
+	Datum	   *invalues;
+	bool	   *inisnull;
+	Datum	   *outvalues;
+	bool	   *outisnull;
+	int			outnatts;
+	int			i;
+
+	/* Sanity checks */
+	Assert(in_slot->tts_tupleDescriptor != NULL &&
+		   out_slot->tts_tupleDescriptor != NULL);
+	Assert(in_slot->tts_values != NULL && out_slot->tts_values != NULL);
+
+	outnatts = out_slot->tts_tupleDescriptor->natts;
+
+	/* Extract all the values of the in slot. */
+	slot_getallattrs(in_slot);
+
+	/* Before doing the mapping, clear any old contents from the out slot */
+	ExecClearTuple(out_slot);
+
+	invalues = in_slot->tts_values;
+	inisnull = in_slot->tts_isnull;
+	outvalues = out_slot->tts_values;
+	outisnull = out_slot->tts_isnull;
+
+	/* Transpose into proper fields of the out slot. */
+	for (i = 0; i < outnatts; i++)
+	{
+		int			j = attrMap[i] - 1;
+
+		/* attrMap[i] == 0 means it's a NULL datum. */
+		if (j == -1)
+		{
+			outvalues[i] = (Datum) 0;
+			outisnull[i] = true;
+		}
+		else
+		{
+			outvalues[i] = invalues[j];
+			outisnull[i] = inisnull[j];
+		}
+	}
+
+	ExecStoreVirtualTuple(out_slot);
+
+	return out_slot;
+}
+
 /*
  * Free a TupleConversionMap structure.
  */

src/backend/commands/analyze.c

@@ -1452,7 +1452,7 @@ acquire_inherited_sample_rows(Relation onerel, int elevel,
 						{
 							HeapTuple	newtup;
 
-							newtup = do_convert_tuple(rows[numrows + j], map);
+							newtup = execute_attr_map_tuple(rows[numrows + j], map);
 							heap_freetuple(rows[numrows + j]);
 							rows[numrows + j] = newtup;
 						}

src/backend/commands/copy.c

@@ -31,6 +31,7 @@
 #include "commands/trigger.h"
 #include "executor/execPartition.h"
 #include "executor/executor.h"
+#include "executor/tuptable.h"
 #include "foreign/fdwapi.h"
 #include "libpq/libpq.h"
 #include "libpq/pqformat.h"
@@ -2691,6 +2692,7 @@ CopyFrom(CopyState cstate)
 		if (proute)
 		{
 			int			leaf_part_index;
+			TupleConversionMap *map;
 
 			/*
 			 * Away we go ... If we end up not finding a partition after all,
@@ -2862,14 +2864,27 @@ CopyFrom(CopyState cstate)
 
 			/*
 			 * We might need to convert from the parent rowtype to the
-			 * partition rowtype.  Don't free the already stored tuple as it
-			 * may still be required for a multi-insert batch.
+			 * partition rowtype.
 			 */
-			tuple = ConvertPartitionTupleSlot(proute->parent_child_tupconv_maps[leaf_part_index],
-											  tuple,
-											  proute->partition_tuple_slot,
-											  &slot,
-											  false);
+			map = proute->parent_child_tupconv_maps[leaf_part_index];
+			if (map != NULL)
+			{
+				TupleTableSlot *new_slot;
+				MemoryContext oldcontext;
+
+				Assert(proute->partition_tuple_slots != NULL &&
+					   proute->partition_tuple_slots[leaf_part_index] != NULL);
+				new_slot = proute->partition_tuple_slots[leaf_part_index];
+				slot = execute_attr_map_slot(map->attrMap, slot, new_slot);
+
+				/*
+				 * Get the tuple in the per-tuple context, so that it will be
+				 * freed after each batch insert.
+				 */
+				oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+				tuple = ExecCopySlotTuple(slot);
+				MemoryContextSwitchTo(oldcontext);
+			}
 
 			tuple->t_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
 		}

src/backend/commands/trigger.c

@@ -5786,7 +5786,7 @@ AfterTriggerSaveEvent(EState *estate, ResultRelInfo *relinfo,
 
 			if (map != NULL)
 			{
-				HeapTuple	converted = do_convert_tuple(oldtup, map);
+				HeapTuple	converted = execute_attr_map_tuple(oldtup, map);
 
 				tuplestore_puttuple(old_tuplestore, converted);
 				pfree(converted);
@@ -5806,7 +5806,7 @@ AfterTriggerSaveEvent(EState *estate, ResultRelInfo *relinfo,
 				tuplestore_puttuple(new_tuplestore, original_insert_tuple);
 			else if (map != NULL)
 			{
-				HeapTuple	converted = do_convert_tuple(newtup, map);
+				HeapTuple	converted = execute_attr_map_tuple(newtup, map);
 
 				tuplestore_puttuple(new_tuplestore, converted);
 				pfree(converted);

src/backend/executor/execExprInterp.c

@@ -3285,7 +3285,7 @@ ExecEvalConvertRowtype(ExprState *state, ExprEvalStep *op, ExprContext *econtext
 	if (op->d.convert_rowtype.map != NULL)
 	{
 		/* Full conversion with attribute rearrangement needed */
-		result = do_convert_tuple(&tmptup, op->d.convert_rowtype.map);
+		result = execute_attr_map_tuple(&tmptup, op->d.convert_rowtype.map);
 		/* Result already has appropriate composite-datum header fields */
 		*op->resvalue = HeapTupleGetDatum(result);
 	}

src/backend/executor/execMain.c

@@ -1955,21 +1955,22 @@ ExecPartitionCheckEmitError(ResultRelInfo *resultRelInfo,
 	 */
 	if (resultRelInfo->ri_PartitionRoot)
 	{
-		HeapTuple	tuple = ExecFetchSlotTuple(slot);
 		TupleDesc	old_tupdesc = RelationGetDescr(rel);
-		TupleConversionMap *map;
+		AttrNumber *map;
 
 		rel = resultRelInfo->ri_PartitionRoot;
 		tupdesc = RelationGetDescr(rel);
 		/* a reverse map */
-		map = convert_tuples_by_name(old_tupdesc, tupdesc,
-									 gettext_noop("could not convert row type"));
+		map = convert_tuples_by_name_map_if_req(old_tupdesc, tupdesc,
+												gettext_noop("could not convert row type"));
+
+		/*
+		 * Partition-specific slot's tupdesc can't be changed, so allocate a
+		 * new one.
+		 */
 		if (map != NULL)
-		{
-			tuple = do_convert_tuple(tuple, map);
-			ExecSetSlotDescriptor(slot, tupdesc);
-			ExecStoreHeapTuple(tuple, slot, false);
-		}
+			slot = execute_attr_map_slot(map, slot,
+										 MakeTupleTableSlot(tupdesc));
 	}
 
 	insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2035,20 +2036,22 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 				 */
 				if (resultRelInfo->ri_PartitionRoot)
 				{
-					HeapTuple	tuple = ExecFetchSlotTuple(slot);
-					TupleConversionMap *map;
+					AttrNumber *map;
 
 					rel = resultRelInfo->ri_PartitionRoot;
 					tupdesc = RelationGetDescr(rel);
 					/* a reverse map */
-					map = convert_tuples_by_name(orig_tupdesc, tupdesc,
-												 gettext_noop("could not convert row type"));
+					map = convert_tuples_by_name_map_if_req(orig_tupdesc,
+															tupdesc,
+															gettext_noop("could not convert row type"));
+
+					/*
+					 * Partition-specific slot's tupdesc can't be changed, so
+					 * allocate a new one.
+					 */
 					if (map != NULL)
-					{
-						tuple = do_convert_tuple(tuple, map);
-						ExecSetSlotDescriptor(slot, tupdesc);
-						ExecStoreHeapTuple(tuple, slot, false);
-					}
+						slot = execute_attr_map_slot(map, slot,
+													 MakeTupleTableSlot(tupdesc));
 				}
 
 				insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2082,21 +2085,23 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 			/* See the comment above. */
 			if (resultRelInfo->ri_PartitionRoot)
 			{
-				HeapTuple	tuple = ExecFetchSlotTuple(slot);
 				TupleDesc	old_tupdesc = RelationGetDescr(rel);
-				TupleConversionMap *map;
+				AttrNumber *map;
 
 				rel = resultRelInfo->ri_PartitionRoot;
 				tupdesc = RelationGetDescr(rel);
 				/* a reverse map */
-				map = convert_tuples_by_name(old_tupdesc, tupdesc,
-											 gettext_noop("could not convert row type"));
+				map = convert_tuples_by_name_map_if_req(old_tupdesc,
+														tupdesc,
+														gettext_noop("could not convert row type"));
+
+				/*
+				 * Partition-specific slot's tupdesc can't be changed, so
+				 * allocate a new one.
+				 */
 				if (map != NULL)
-				{
-					tuple = do_convert_tuple(tuple, map);
-					ExecSetSlotDescriptor(slot, tupdesc);
-					ExecStoreHeapTuple(tuple, slot, false);
-				}
+					slot = execute_attr_map_slot(map, slot,
+												 MakeTupleTableSlot(tupdesc));
 			}
 
 			insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2188,21 +2193,23 @@ ExecWithCheckOptions(WCOKind kind, ResultRelInfo *resultRelInfo,
 					/* See the comment in ExecConstraints(). */
 					if (resultRelInfo->ri_PartitionRoot)
 					{
-						HeapTuple	tuple = ExecFetchSlotTuple(slot);
 						TupleDesc	old_tupdesc = RelationGetDescr(rel);
-						TupleConversionMap *map;
+						AttrNumber *map;
 
 						rel = resultRelInfo->ri_PartitionRoot;
 						tupdesc = RelationGetDescr(rel);
 						/* a reverse map */
-						map = convert_tuples_by_name(old_tupdesc, tupdesc,
-													 gettext_noop("could not convert row type"));
+						map = convert_tuples_by_name_map_if_req(old_tupdesc,
+																tupdesc,
+																gettext_noop("could not convert row type"));
+
+						/*
+						 * Partition-specific slot's tupdesc can't be changed,
+						 * so allocate a new one.
+						 */
 						if (map != NULL)
-						{
-							tuple = do_convert_tuple(tuple, map);
-							ExecSetSlotDescriptor(slot, tupdesc);
-							ExecStoreHeapTuple(tuple, slot, false);
-						}
+							slot = execute_attr_map_slot(map, slot,
+														 MakeTupleTableSlot(tupdesc));
 					}
 
 					insertedCols = GetInsertedColumns(resultRelInfo, estate);

src/backend/executor/execPartition.c

@@ -57,7 +57,7 @@ typedef struct PartitionDispatchData
 	List	   *keystate;		/* list of ExprState */
 	PartitionDesc partdesc;
 	TupleTableSlot *tupslot;
-	TupleConversionMap *tupmap;
+	AttrNumber *tupmap;
 	int		   *indexes;
 } PartitionDispatchData;
 
@@ -144,17 +144,9 @@ ExecSetupPartitionTupleRouting(ModifyTableState *mtstate, Relation rel)
 		 * We need an additional tuple slot for storing transient tuples that
 		 * are converted to the root table descriptor.
 		 */
-		proute->root_tuple_slot = MakeTupleTableSlot(NULL);
+		proute->root_tuple_slot = MakeTupleTableSlot(RelationGetDescr(rel));
 	}
 
-	/*
-	 * Initialize an empty slot that will be used to manipulate tuples of any
-	 * given partition's rowtype.  It is attached to the caller-specified node
-	 * (such as ModifyTableState) and released when the node finishes
-	 * processing.
-	 */
-	proute->partition_tuple_slot = MakeTupleTableSlot(NULL);
-
 	i = 0;
 	foreach(cell, leaf_parts)
 	{
@@ -228,7 +220,6 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
 	TupleTableSlot *ecxt_scantuple_old = ecxt->ecxt_scantuple;
 	TupleTableSlot *myslot = NULL;
 	MemoryContext	oldcxt;
-	HeapTuple		tuple;
 
 	/* use per-tuple context here to avoid leaking memory */
 	oldcxt = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
@@ -241,11 +232,10 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
 		ExecPartitionCheck(resultRelInfo, slot, estate, true);
 
 	/* start with the root partitioned table */
-	tuple = ExecFetchSlotTuple(slot);
 	dispatch = pd[0];
 	while (true)
 	{
-		TupleConversionMap *map = dispatch->tupmap;
+		AttrNumber *map = dispatch->tupmap;
 		int			cur_index = -1;
 
 		rel = dispatch->reldesc;
@@ -256,11 +246,7 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
 		 */
 		myslot = dispatch->tupslot;
 		if (myslot != NULL && map != NULL)
-		{
-			tuple = do_convert_tuple(tuple, map);
-			ExecStoreHeapTuple(tuple, myslot, true);
-			slot = myslot;
-		}
+			slot = execute_attr_map_slot(map, slot, myslot);
 
 		/*
 		 * Extract partition key from tuple. Expression evaluation machinery
@@ -305,16 +291,6 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
 		{
 			/* move down one level */
 			dispatch = pd[-dispatch->indexes[cur_index]];
-
-			/*
-			 * Release the dedicated slot, if it was used.  Create a copy of
-			 * the tuple first, for the next iteration.
-			 */
-			if (slot == myslot)
-			{
-				tuple = ExecCopySlotTuple(myslot);
-				ExecClearTuple(myslot);
-			}
 		}
 	}
 
@@ -738,6 +714,35 @@ ExecInitRoutingInfo(ModifyTableState *mtstate,
 							   RelationGetDescr(partRelInfo->ri_RelationDesc),
 							   gettext_noop("could not convert row type"));
 
+	/*
+	 * If a partition has a different rowtype than the root parent, initialize
+	 * a slot dedicated to storing this partition's tuples.  The slot is used
+	 * for various operations that are applied to tuples after routing, such
+	 * as checking constraints.
+	 */
+	if (proute->parent_child_tupconv_maps[partidx] != NULL)
+	{
+		Relation	partrel = partRelInfo->ri_RelationDesc;
+
+		/*
+		 * Initialize the array in proute where these slots are stored, if not
+		 * already done.
+		 */
+		if (proute->partition_tuple_slots == NULL)
+			proute->partition_tuple_slots = (TupleTableSlot **)
+				palloc0(proute->num_partitions *
+						sizeof(TupleTableSlot *));
+
+		/*
+		 * Initialize the slot itself setting its descriptor to this
+		 * partition's TupleDesc; TupleDesc reference will be released at the
+		 * end of the command.
+		 */
+		proute->partition_tuple_slots[partidx] =
+			ExecInitExtraTupleSlot(estate,
+								   RelationGetDescr(partrel));
+	}
+
 	/*
 	 * If the partition is a foreign table, let the FDW init itself for
 	 * routing tuples to the partition.
@@ -815,38 +820,6 @@ TupConvMapForLeaf(PartitionTupleRouting *proute,
 	return *map;
 }
 
-/*
- * ConvertPartitionTupleSlot -- convenience function for tuple conversion.
- * The tuple, if converted, is stored in new_slot, and *p_my_slot is
- * updated to point to it.  new_slot typically should be one of the
- * dedicated partition tuple slots. If map is NULL, *p_my_slot is not changed.
- *
- * Returns the converted tuple, unless map is NULL, in which case original
- * tuple is returned unmodified.
- */
-HeapTuple
-ConvertPartitionTupleSlot(TupleConversionMap *map,
-						  HeapTuple tuple,
-						  TupleTableSlot *new_slot,
-						  TupleTableSlot **p_my_slot,
-						  bool shouldFree)
-{
-	if (!map)
-		return tuple;
-
-	tuple = do_convert_tuple(tuple, map);
-
-	/*
-	 * Change the partition tuple slot descriptor, as per converted tuple.
-	 */
-	*p_my_slot = new_slot;
-	Assert(new_slot != NULL);
-	ExecSetSlotDescriptor(new_slot, map->outdesc);
-	ExecStoreHeapTuple(tuple, new_slot, shouldFree);
-
-	return tuple;
-}
-
 /*
  * ExecCleanupTupleRouting -- Clean up objects allocated for partition tuple
  * routing.
@@ -915,8 +888,6 @@ ExecCleanupTupleRouting(ModifyTableState *mtstate,
 	/* Release the standalone partition tuple descriptors, if any */
 	if (proute->root_tuple_slot)
 		ExecDropSingleTupleTableSlot(proute->root_tuple_slot);
-	if (proute->partition_tuple_slot)
-		ExecDropSingleTupleTableSlot(proute->partition_tuple_slot);
 }
 
 /*
@@ -1004,9 +975,9 @@ get_partition_dispatch_recurse(Relation rel, Relation parent,
 		 * for tuple routing.
 		 */
 		pd->tupslot = MakeSingleTupleTableSlot(tupdesc);
-		pd->tupmap = convert_tuples_by_name(RelationGetDescr(parent),
-											tupdesc,
-											gettext_noop("could not convert row type"));
+		pd->tupmap = convert_tuples_by_name_map_if_req(RelationGetDescr(parent),
+													   tupdesc,
+													   gettext_noop("could not convert row type"));
 	}
 	else
 	{

src/backend/executor/nodeModifyTable.c

@@ -1161,11 +1161,9 @@ lreplace:;
 			map_index = resultRelInfo - mtstate->resultRelInfo;
 			Assert(map_index >= 0 && map_index < mtstate->mt_nplans);
 			tupconv_map = tupconv_map_for_subplan(mtstate, map_index);
-			tuple = ConvertPartitionTupleSlot(tupconv_map,
-											  tuple,
-											  proute->root_tuple_slot,
-											  &slot,
-											  true);
+			if (tupconv_map != NULL)
+				slot = execute_attr_map_slot(tupconv_map->attrMap,
+											 slot, proute->root_tuple_slot);
 
 			/*
 			 * Prepare for tuple routing, making it look like we're inserting
@@ -1703,6 +1701,7 @@ ExecPrepareTupleRouting(ModifyTableState *mtstate,
 	int			partidx;
 	ResultRelInfo *partrel;
 	HeapTuple	tuple;
+	TupleConversionMap *map;
 
 	/*
 	 * Determine the target partition.  If ExecFindPartition does not find a
@@ -1790,11 +1789,16 @@ ExecPrepareTupleRouting(ModifyTableState *mtstate,
 	/*
 	 * Convert the tuple, if necessary.
 	 */
-	ConvertPartitionTupleSlot(proute->parent_child_tupconv_maps[partidx],
-							  tuple,
-							  proute->partition_tuple_slot,
-							  &slot,
-							  true);
+	map = proute->parent_child_tupconv_maps[partidx];
+	if (map != NULL)
+	{
+		TupleTableSlot *new_slot;
+
+		Assert(proute->partition_tuple_slots != NULL &&
+			   proute->partition_tuple_slots[partidx] != NULL);
+		new_slot = proute->partition_tuple_slots[partidx];
+		slot = execute_attr_map_slot(map->attrMap, slot, new_slot);
+	}
 
 	/* Initialize information needed to handle ON CONFLICT DO UPDATE. */
 	Assert(mtstate != NULL);

src/include/access/tupconvert.h

@@ -16,6 +16,7 @@
 
 #include "access/htup.h"
 #include "access/tupdesc.h"
+#include "executor/tuptable.h"
 
 
 typedef struct TupleConversionMap
@@ -41,8 +42,13 @@ extern TupleConversionMap *convert_tuples_by_name(TupleDesc indesc,
 extern AttrNumber *convert_tuples_by_name_map(TupleDesc indesc,
 						   TupleDesc outdesc,
 						   const char *msg);
+extern AttrNumber *convert_tuples_by_name_map_if_req(TupleDesc indesc,
+								  TupleDesc outdesc,
+								  const char *msg);
 
-extern HeapTuple do_convert_tuple(HeapTuple tuple, TupleConversionMap *map);
+extern HeapTuple execute_attr_map_tuple(HeapTuple tuple, TupleConversionMap *map);
+extern TupleTableSlot *execute_attr_map_slot(AttrNumber *attrMap,
+					  TupleTableSlot *in_slot, TupleTableSlot *out_slot);
 
 extern void free_conversion_map(TupleConversionMap *map);
 

src/include/executor/execPartition.h

@@ -58,10 +58,13 @@ typedef struct PartitionDispatchData *PartitionDispatch;
  *								element of this array has the index into the
  *								corresponding partition in partitions array.
  * num_subplan_partition_offsets  Length of 'subplan_partition_offsets' array
- * partition_tuple_slot			TupleTableSlot to be used to manipulate any
- *								given leaf partition's rowtype after that
- *								partition is chosen for insertion by
- *								tuple-routing.
+ * partition_tuple_slots		Array of TupleTableSlot objects; if non-NULL,
+ *								contains one entry for every leaf partition,
+ *								of which only those of the leaf partitions
+ *								whose attribute numbers differ from the root
+ *								parent have a non-NULL value.  NULL if all of
+ *								the partitions encountered by a given command
+ *								happen to have same rowtype as the root parent
  * root_tuple_slot				TupleTableSlot to be used to transiently hold
  *								copy of a tuple that's being moved across
  *								partitions in the root partitioned table's
@@ -80,7 +83,7 @@ typedef struct PartitionTupleRouting
 	bool	   *child_parent_map_not_required;
 	int		   *subplan_partition_offsets;
 	int			num_subplan_partition_offsets;
-	TupleTableSlot *partition_tuple_slot;
+	TupleTableSlot **partition_tuple_slots;
 	TupleTableSlot *root_tuple_slot;
 } PartitionTupleRouting;
 
@@ -188,11 +191,6 @@ extern void ExecInitRoutingInfo(ModifyTableState *mtstate,
 extern void ExecSetupChildParentMapForLeaf(PartitionTupleRouting *proute);
 extern TupleConversionMap *TupConvMapForLeaf(PartitionTupleRouting *proute,
 				  ResultRelInfo *rootRelInfo, int leaf_index);
-extern HeapTuple ConvertPartitionTupleSlot(TupleConversionMap *map,
-						  HeapTuple tuple,
-						  TupleTableSlot *new_slot,
-						  TupleTableSlot **p_my_slot,
-						  bool shouldFree);
 extern void ExecCleanupTupleRouting(ModifyTableState *mtstate,
 						PartitionTupleRouting *proute);
 extern PartitionPruneState *ExecCreatePartitionPruneState(PlanState *planstate,

src/pl/plpgsql/src/pl_exec.c

@@ -798,7 +798,7 @@ coerce_function_result_tuple(PLpgSQL_execstate *estate, TupleDesc tupdesc)
 		{
 			rettup = expanded_record_get_tuple(erh);
 			Assert(rettup);
-			rettup = do_convert_tuple(rettup, tupmap);
+			rettup = execute_attr_map_tuple(rettup, tupmap);
 
 			/*
 			 * Copy tuple to upper executor memory, as a tuple Datum.  Make
@@ -834,7 +834,7 @@ coerce_function_result_tuple(PLpgSQL_execstate *estate, TupleDesc tupdesc)
 
 		/* it might need conversion */
 		if (tupmap)
-			rettup = do_convert_tuple(rettup, tupmap);
+			rettup = execute_attr_map_tuple(rettup, tupmap);
 
 		/*
 		 * Copy tuple to upper executor memory, as a tuple Datum.  Make sure
@@ -1011,7 +1011,7 @@ plpgsql_exec_trigger(PLpgSQL_function *func,
 													gettext_noop("returned row structure does not match the structure of the triggering table"));
 				/* it might need conversion */
 				if (tupmap)
-					rettup = do_convert_tuple(rettup, tupmap);
+					rettup = execute_attr_map_tuple(rettup, tupmap);
 				/* no need to free map, we're about to return anyway */
 			}
 
@@ -1039,7 +1039,7 @@ plpgsql_exec_trigger(PLpgSQL_function *func,
 												gettext_noop("returned row structure does not match the structure of the triggering table"));
 			/* it might need conversion */
 			if (tupmap)
-				rettup = do_convert_tuple(rettup, tupmap);
+				rettup = execute_attr_map_tuple(rettup, tupmap);
 
 			ReleaseTupleDesc(retdesc);
 			/* no need to free map, we're about to return anyway */
@@ -3292,7 +3292,7 @@ exec_stmt_return_next(PLpgSQL_execstate *estate,
 														gettext_noop("wrong record type supplied in RETURN NEXT"));
 					tuple = expanded_record_get_tuple(rec->erh);
 					if (tupmap)
-						tuple = do_convert_tuple(tuple, tupmap);
+						tuple = execute_attr_map_tuple(tuple, tupmap);
 					tuplestore_puttuple(estate->tuple_store, tuple);
 					MemoryContextSwitchTo(oldcontext);
 				}
@@ -3355,7 +3355,7 @@ exec_stmt_return_next(PLpgSQL_execstate *estate,
 				tupmap = convert_tuples_by_position(retvaldesc, tupdesc,
 													gettext_noop("returned record type does not match expected record type"));
 				if (tupmap)
-					tuple = do_convert_tuple(tuple, tupmap);
+					tuple = execute_attr_map_tuple(tuple, tupmap);
 				tuplestore_puttuple(estate->tuple_store, tuple);
 				ReleaseTupleDesc(retvaldesc);
 				MemoryContextSwitchTo(oldcontext);
@@ -3471,7 +3471,7 @@ exec_stmt_return_query(PLpgSQL_execstate *estate,
 			HeapTuple	tuple = SPI_tuptable->vals[i];
 
 			if (tupmap)
-				tuple = do_convert_tuple(tuple, tupmap);
+				tuple = execute_attr_map_tuple(tuple, tupmap);
 			tuplestore_puttuple(estate->tuple_store, tuple);
 			if (tupmap)
 				heap_freetuple(tuple);