Slotification of partition tuple conversion

diff --git a/src/backend/access/common/tupconvert.c b/src/backend/access/common/tupconvert.c
index 3bc67b8..fbcfa85 100644
--- a/src/backend/access/common/tupconvert.c
+++ b/src/backend/access/common/tupconvert.c
@@ -22,6 +22,7 @@
 
 #include "access/htup_details.h"
 #include "access/tupconvert.h"
+#include "executor/tuptable.h"
 #include "utils/builtins.h"
 
 
@@ -406,6 +407,60 @@ do_convert_tuple(HeapTuple tuple, TupleConversionMap *map)
 }
 
 /*
+ * Perform conversion of a tuple slot according to the map.
+ */
+TupleTableSlot *
+ConvertTupleSlot(TupleConversionMap *map,
+				 TupleTableSlot *in_slot, TupleTableSlot *out_slot)
+{
+	AttrNumber *attrMap = map->attrMap;
+	Datum	   *invalues;
+	bool	   *inisnull;
+	Datum	   *outvalues;
+	bool	   *outisnull;
+	int			outnatts = map->outdesc->natts;
+	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);
+
+	/* 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.
  */
 void
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index f61db3e..9084e9c 100644
--- a/src/backend/commands/copy.c
+++ b/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"
@@ -2869,12 +2870,21 @@ CopyFrom(CopyState cstate)
 			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];
-				tuple = ConvertPartitionTupleSlot(map, tuple, new_slot, &slot,
-												  false);
+				slot = ConvertTupleSlot(map, slot, new_slot);
+
+				/*
+				 * Get the tuple in the per-tuple context. Also, we cannot call
+				 * ExecMaterializeSlot(), otherwise the tuple will get freed
+				 * while storing the next tuple.
+				 */
+				oldcontext = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+				tuple = ExecCopySlotTuple(slot);
+				MemoryContextSwitchTo(oldcontext);
 			}
 
 			tuple->t_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 415670e..ecdf7f2 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1927,7 +1927,6 @@ ExecPartitionCheckEmitError(ResultRelInfo *resultRelInfo,
 	 */
 	if (resultRelInfo->ri_PartitionRoot)
 	{
-		HeapTuple	tuple = ExecFetchSlotTuple(slot);
 		TupleDesc	old_tupdesc = RelationGetDescr(rel);
 		TupleConversionMap *map;
 
@@ -1936,16 +1935,12 @@ ExecPartitionCheckEmitError(ResultRelInfo *resultRelInfo,
 		/* a reverse map */
 		map = convert_tuples_by_name(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)
-		{
-			/*
-			 * Partition-specific slot's tupdesc can't be changed, so allocate
-			 * a new one.
-			 */
-			slot = MakeTupleTableSlot(tupdesc);
-			tuple = do_convert_tuple(tuple, map);
-			ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-		}
+			slot = ConvertTupleSlot(map, slot, MakeTupleTableSlot(tupdesc));
 	}
 
 	insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2011,7 +2006,6 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 				 */
 				if (resultRelInfo->ri_PartitionRoot)
 				{
-					HeapTuple	tuple = ExecFetchSlotTuple(slot);
 					TupleConversionMap *map;
 
 					rel = resultRelInfo->ri_PartitionRoot;
@@ -2019,16 +2013,13 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 					/* a reverse map */
 					map = convert_tuples_by_name(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)
-					{
-						/*
-						 * Partition-specific slot's tupdesc can't be changed,
-						 * so allocate a new one.
-						 */
-						slot = MakeTupleTableSlot(tupdesc);
-						tuple = do_convert_tuple(tuple, map);
-						ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-					}
+						slot = ConvertTupleSlot(map, slot,
+												MakeTupleTableSlot(tupdesc));
 				}
 
 				insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2062,7 +2053,6 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 			/* See the comment above. */
 			if (resultRelInfo->ri_PartitionRoot)
 			{
-				HeapTuple	tuple = ExecFetchSlotTuple(slot);
 				TupleDesc	old_tupdesc = RelationGetDescr(rel);
 				TupleConversionMap *map;
 
@@ -2071,16 +2061,13 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 				/* a reverse map */
 				map = convert_tuples_by_name(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)
-				{
-					/*
-					 * Partition-specific slot's tupdesc can't be changed,
-					 * so allocate a new one.
-					 */
-					slot = MakeTupleTableSlot(tupdesc);
-					tuple = do_convert_tuple(tuple, map);
-					ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-				}
+					slot = ConvertTupleSlot(map, slot,
+											MakeTupleTableSlot(tupdesc));
 			}
 
 			insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2172,7 +2159,6 @@ 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;
 
@@ -2181,16 +2167,13 @@ ExecWithCheckOptions(WCOKind kind, ResultRelInfo *resultRelInfo,
 						/* a reverse map */
 						map = convert_tuples_by_name(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)
-						{
-							/*
-							 * Partition-specific slot's tupdesc can't be
-							 * changed, so allocate a new one.
-							 */
-							slot = MakeTupleTableSlot(tupdesc);
-							tuple = do_convert_tuple(tuple, map);
-							ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-						}
+							slot = ConvertTupleSlot(map, slot,
+													MakeTupleTableSlot(tupdesc));
 					}
 
 					insertedCols = GetInsertedColumns(resultRelInfo, estate);
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index 2af7599..6c9b8f9 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -190,7 +190,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));
@@ -203,7 +202,6 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
 		ExecPartitionCheck(resultRelInfo, slot, estate, true);
 
 	/* start with the root partitioned table */
-	tuple = ExecFetchSlotTuple(slot);
 	dispatch = pd[0];
 	while (true)
 	{
@@ -218,11 +216,7 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
 		 */
 		myslot = dispatch->tupslot;
 		if (myslot != NULL && map != NULL)
-		{
-			tuple = do_convert_tuple(tuple, map);
-			ExecStoreTuple(tuple, myslot, InvalidBuffer, true);
-			slot = myslot;
-		}
+			slot = ConvertTupleSlot(map, slot, myslot);
 
 		/*
 		 * Extract partition key from tuple. Expression evaluation machinery
@@ -267,16 +261,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);
-			}
 		}
 	}
 
@@ -807,36 +791,6 @@ TupConvMapForLeaf(PartitionTupleRouting *proute,
 }
 
 /*
- * 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)
-{
-	Assert(map != NULL && new_slot != NULL);
-
-	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);
-	ExecStoreTuple(tuple, new_slot, InvalidBuffer, shouldFree);
-
-	return tuple;
-}
-
-/*
  * ExecCleanupTupleRouting -- Clean up objects allocated for partition tuple
  * routing.
  *
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index e4b2d37..65cd53a 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -1162,11 +1162,8 @@ lreplace:;
 			Assert(map_index >= 0 && map_index < mtstate->mt_nplans);
 			tupconv_map = tupconv_map_for_subplan(mtstate, map_index);
 			if (tupconv_map != NULL)
-				tuple = ConvertPartitionTupleSlot(tupconv_map,
-												  tuple,
-												  proute->root_tuple_slot,
-												  &slot,
-												  true);
+				slot = ConvertTupleSlot(tupconv_map,
+										slot, proute->root_tuple_slot);
 
 			/*
 			 * Prepare for tuple routing, making it look like we're inserting
@@ -1800,7 +1797,7 @@ ExecPrepareTupleRouting(ModifyTableState *mtstate,
 		Assert(proute->partition_tuple_slots != NULL &&
 			   proute->partition_tuple_slots[partidx] != NULL);
 		new_slot = proute->partition_tuple_slots[partidx];
-		ConvertPartitionTupleSlot(map, tuple, new_slot, &slot, true);
+		slot = ConvertTupleSlot(map, slot, new_slot);
 	}
 
 	/* Initialize information needed to handle ON CONFLICT DO UPDATE. */
diff --git a/src/include/access/tupconvert.h b/src/include/access/tupconvert.h
index 66c0ed0..7a453fe 100644
--- a/src/include/access/tupconvert.h
+++ b/src/include/access/tupconvert.h
@@ -16,6 +16,7 @@
 
 #include "access/htup.h"
 #include "access/tupdesc.h"
+#include "executor/tuptable.h"
 
 
 typedef struct TupleConversionMap
@@ -44,6 +45,9 @@ extern AttrNumber *convert_tuples_by_name_map(TupleDesc indesc,
 
 extern HeapTuple do_convert_tuple(HeapTuple tuple, TupleConversionMap *map);
 
+extern TupleTableSlot *ConvertTupleSlot(TupleConversionMap *map,
+				 TupleTableSlot *in_slot, TupleTableSlot *out_slot);
+
 extern void free_conversion_map(TupleConversionMap *map);
 
 #endif							/* TUPCONVERT_H */
diff --git a/src/include/executor/execPartition.h b/src/include/executor/execPartition.h
index c2a2dc5..13a213e 100644
--- a/src/include/executor/execPartition.h
+++ b/src/include/executor/execPartition.h
@@ -219,11 +219,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,

diff --git a/src/backend/access/common/tupconvert.c b/src/backend/access/common/tupconvert.c
index 3bc67b8..fbcfa85 100644
--- a/src/backend/access/common/tupconvert.c
+++ b/src/backend/access/common/tupconvert.c
@@ -22,6 +22,7 @@
 
 #include "access/htup_details.h"
 #include "access/tupconvert.h"
+#include "executor/tuptable.h"
 #include "utils/builtins.h"
 
 
@@ -406,6 +407,60 @@ do_convert_tuple(HeapTuple tuple, TupleConversionMap *map)
 }
 
 /*
+ * Perform conversion of a tuple slot according to the map.
+ */
+TupleTableSlot *
+ConvertTupleSlot(TupleConversionMap *map,
+				 TupleTableSlot *in_slot, TupleTableSlot *out_slot)
+{
+	AttrNumber *attrMap = map->attrMap;
+	Datum	   *invalues;
+	bool	   *inisnull;
+	Datum	   *outvalues;
+	bool	   *outisnull;
+	int			outnatts = map->outdesc->natts;
+	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);
+
+	/* 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.
  */
 void
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index f61db3e..e5e8e20 100644
--- a/src/backend/commands/copy.c
+++ b/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"
@@ -2869,12 +2870,20 @@ CopyFrom(CopyState cstate)
 			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];
-				tuple = ConvertPartitionTupleSlot(map, tuple, new_slot, &slot,
-												  false);
+				slot = ConvertTupleSlot(map, 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);
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 415670e..ecdf7f2 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1927,7 +1927,6 @@ ExecPartitionCheckEmitError(ResultRelInfo *resultRelInfo,
 	 */
 	if (resultRelInfo->ri_PartitionRoot)
 	{
-		HeapTuple	tuple = ExecFetchSlotTuple(slot);
 		TupleDesc	old_tupdesc = RelationGetDescr(rel);
 		TupleConversionMap *map;
 
@@ -1936,16 +1935,12 @@ ExecPartitionCheckEmitError(ResultRelInfo *resultRelInfo,
 		/* a reverse map */
 		map = convert_tuples_by_name(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)
-		{
-			/*
-			 * Partition-specific slot's tupdesc can't be changed, so allocate
-			 * a new one.
-			 */
-			slot = MakeTupleTableSlot(tupdesc);
-			tuple = do_convert_tuple(tuple, map);
-			ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-		}
+			slot = ConvertTupleSlot(map, slot, MakeTupleTableSlot(tupdesc));
 	}
 
 	insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2011,7 +2006,6 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 				 */
 				if (resultRelInfo->ri_PartitionRoot)
 				{
-					HeapTuple	tuple = ExecFetchSlotTuple(slot);
 					TupleConversionMap *map;
 
 					rel = resultRelInfo->ri_PartitionRoot;
@@ -2019,16 +2013,13 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 					/* a reverse map */
 					map = convert_tuples_by_name(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)
-					{
-						/*
-						 * Partition-specific slot's tupdesc can't be changed,
-						 * so allocate a new one.
-						 */
-						slot = MakeTupleTableSlot(tupdesc);
-						tuple = do_convert_tuple(tuple, map);
-						ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-					}
+						slot = ConvertTupleSlot(map, slot,
+												MakeTupleTableSlot(tupdesc));
 				}
 
 				insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2062,7 +2053,6 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 			/* See the comment above. */
 			if (resultRelInfo->ri_PartitionRoot)
 			{
-				HeapTuple	tuple = ExecFetchSlotTuple(slot);
 				TupleDesc	old_tupdesc = RelationGetDescr(rel);
 				TupleConversionMap *map;
 
@@ -2071,16 +2061,13 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 				/* a reverse map */
 				map = convert_tuples_by_name(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)
-				{
-					/*
-					 * Partition-specific slot's tupdesc can't be changed,
-					 * so allocate a new one.
-					 */
-					slot = MakeTupleTableSlot(tupdesc);
-					tuple = do_convert_tuple(tuple, map);
-					ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-				}
+					slot = ConvertTupleSlot(map, slot,
+											MakeTupleTableSlot(tupdesc));
 			}
 
 			insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2172,7 +2159,6 @@ 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;
 
@@ -2181,16 +2167,13 @@ ExecWithCheckOptions(WCOKind kind, ResultRelInfo *resultRelInfo,
 						/* a reverse map */
 						map = convert_tuples_by_name(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)
-						{
-							/*
-							 * Partition-specific slot's tupdesc can't be
-							 * changed, so allocate a new one.
-							 */
-							slot = MakeTupleTableSlot(tupdesc);
-							tuple = do_convert_tuple(tuple, map);
-							ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-						}
+							slot = ConvertTupleSlot(map, slot,
+													MakeTupleTableSlot(tupdesc));
 					}
 
 					insertedCols = GetInsertedColumns(resultRelInfo, estate);
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index 2af7599..6c9b8f9 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -190,7 +190,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));
@@ -203,7 +202,6 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
 		ExecPartitionCheck(resultRelInfo, slot, estate, true);
 
 	/* start with the root partitioned table */
-	tuple = ExecFetchSlotTuple(slot);
 	dispatch = pd[0];
 	while (true)
 	{
@@ -218,11 +216,7 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
 		 */
 		myslot = dispatch->tupslot;
 		if (myslot != NULL && map != NULL)
-		{
-			tuple = do_convert_tuple(tuple, map);
-			ExecStoreTuple(tuple, myslot, InvalidBuffer, true);
-			slot = myslot;
-		}
+			slot = ConvertTupleSlot(map, slot, myslot);
 
 		/*
 		 * Extract partition key from tuple. Expression evaluation machinery
@@ -267,16 +261,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);
-			}
 		}
 	}
 
@@ -807,36 +791,6 @@ TupConvMapForLeaf(PartitionTupleRouting *proute,
 }
 
 /*
- * 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)
-{
-	Assert(map != NULL && new_slot != NULL);
-
-	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);
-	ExecStoreTuple(tuple, new_slot, InvalidBuffer, shouldFree);
-
-	return tuple;
-}
-
-/*
  * ExecCleanupTupleRouting -- Clean up objects allocated for partition tuple
  * routing.
  *
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index e4b2d37..65cd53a 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -1162,11 +1162,8 @@ lreplace:;
 			Assert(map_index >= 0 && map_index < mtstate->mt_nplans);
 			tupconv_map = tupconv_map_for_subplan(mtstate, map_index);
 			if (tupconv_map != NULL)
-				tuple = ConvertPartitionTupleSlot(tupconv_map,
-												  tuple,
-												  proute->root_tuple_slot,
-												  &slot,
-												  true);
+				slot = ConvertTupleSlot(tupconv_map,
+										slot, proute->root_tuple_slot);
 
 			/*
 			 * Prepare for tuple routing, making it look like we're inserting
@@ -1800,7 +1797,7 @@ ExecPrepareTupleRouting(ModifyTableState *mtstate,
 		Assert(proute->partition_tuple_slots != NULL &&
 			   proute->partition_tuple_slots[partidx] != NULL);
 		new_slot = proute->partition_tuple_slots[partidx];
-		ConvertPartitionTupleSlot(map, tuple, new_slot, &slot, true);
+		slot = ConvertTupleSlot(map, slot, new_slot);
 	}
 
 	/* Initialize information needed to handle ON CONFLICT DO UPDATE. */
diff --git a/src/include/access/tupconvert.h b/src/include/access/tupconvert.h
index 66c0ed0..7a453fe 100644
--- a/src/include/access/tupconvert.h
+++ b/src/include/access/tupconvert.h
@@ -16,6 +16,7 @@
 
 #include "access/htup.h"
 #include "access/tupdesc.h"
+#include "executor/tuptable.h"
 
 
 typedef struct TupleConversionMap
@@ -44,6 +45,9 @@ extern AttrNumber *convert_tuples_by_name_map(TupleDesc indesc,
 
 extern HeapTuple do_convert_tuple(HeapTuple tuple, TupleConversionMap *map);
 
+extern TupleTableSlot *ConvertTupleSlot(TupleConversionMap *map,
+				 TupleTableSlot *in_slot, TupleTableSlot *out_slot);
+
 extern void free_conversion_map(TupleConversionMap *map);
 
 #endif							/* TUPCONVERT_H */
diff --git a/src/include/executor/execPartition.h b/src/include/executor/execPartition.h
index c2a2dc5..13a213e 100644
--- a/src/include/executor/execPartition.h
+++ b/src/include/executor/execPartition.h
@@ -219,11 +219,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,

diff --git a/src/backend/access/common/tupconvert.c b/src/backend/access/common/tupconvert.c
index 3bc67b8..2100d40 100644
--- a/src/backend/access/common/tupconvert.c
+++ b/src/backend/access/common/tupconvert.c
@@ -22,6 +22,7 @@
 
 #include "access/htup_details.h"
 #include "access/tupconvert.h"
+#include "executor/tuptable.h"
 #include "utils/builtins.h"
 
 
@@ -214,6 +215,45 @@ convert_tuples_by_name(TupleDesc indesc,
 	TupleConversionMap *map;
 	AttrNumber *attrMap;
 	int			n = outdesc->natts;
+
+	/* Verify compatibility and prepare attribute-number map */
+	attrMap = inheritence_tupconv_map(indesc, outdesc, msg);
+
+	if (attrMap == NULL)
+	{
+		/* Runtime conversion is not needed */
+		return NULL;
+	}
+
+	/* Prepare the map structure */
+	map = (TupleConversionMap *) palloc(sizeof(TupleConversionMap));
+	map->indesc = indesc;
+	map->outdesc = outdesc;
+	map->attrMap = attrMap;
+	/* preallocate workspace for Datum arrays */
+	map->outvalues = (Datum *) palloc(n * sizeof(Datum));
+	map->outisnull = (bool *) palloc(n * sizeof(bool));
+	n = indesc->natts + 1;		/* +1 for NULL */
+	map->invalues = (Datum *) palloc(n * sizeof(Datum));
+	map->inisnull = (bool *) palloc(n * sizeof(bool));
+	map->invalues[0] = (Datum) 0;	/* set up the NULL entry */
+	map->inisnull[0] = true;
+
+	return map;
+}
+
+/*
+ * Returns a bare attribute map for tuple conversion. Returns NULL if
+ * conversion not required. This is a convenience routine for
+ * convert_tuples_by_name() and other functions.
+ */
+AttrNumber *
+inheritence_tupconv_map(TupleDesc indesc,
+						TupleDesc outdesc,
+						const char *msg)
+{
+	AttrNumber *attrMap;
+	int			n = outdesc->natts;
 	int			i;
 	bool		same;
 
@@ -265,22 +305,8 @@ convert_tuples_by_name(TupleDesc indesc,
 		pfree(attrMap);
 		return NULL;
 	}
-
-	/* Prepare the map structure */
-	map = (TupleConversionMap *) palloc(sizeof(TupleConversionMap));
-	map->indesc = indesc;
-	map->outdesc = outdesc;
-	map->attrMap = attrMap;
-	/* preallocate workspace for Datum arrays */
-	map->outvalues = (Datum *) palloc(n * sizeof(Datum));
-	map->outisnull = (bool *) palloc(n * sizeof(bool));
-	n = indesc->natts + 1;		/* +1 for NULL */
-	map->invalues = (Datum *) palloc(n * sizeof(Datum));
-	map->inisnull = (bool *) palloc(n * sizeof(bool));
-	map->invalues[0] = (Datum) 0;	/* set up the NULL entry */
-	map->inisnull[0] = true;
-
-	return map;
+	else
+		return attrMap;
 }
 
 /*
@@ -406,6 +432,61 @@ do_convert_tuple(HeapTuple tuple, TupleConversionMap *map)
 }
 
 /*
+ * Perform conversion of a tuple slot according to the map.
+ */
+TupleTableSlot *
+ConvertTupleSlot(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.
  */
 void
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index f61db3e..a4085a4 100644
--- a/src/backend/commands/copy.c
+++ b/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"
@@ -2862,19 +2863,26 @@ 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.
 			 */
 			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];
-				tuple = ConvertPartitionTupleSlot(map, tuple, new_slot, &slot,
-												  false);
+				slot = ConvertTupleSlot(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);
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 415670e..f696231 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1927,25 +1927,20 @@ 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 = inheritence_tupconv_map(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)
-		{
-			/*
-			 * Partition-specific slot's tupdesc can't be changed, so allocate
-			 * a new one.
-			 */
-			slot = MakeTupleTableSlot(tupdesc);
-			tuple = do_convert_tuple(tuple, map);
-			ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-		}
+			slot = ConvertTupleSlot(map, slot, MakeTupleTableSlot(tupdesc));
 	}
 
 	insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2011,24 +2006,20 @@ 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 = inheritence_tupconv_map(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)
-					{
-						/*
-						 * Partition-specific slot's tupdesc can't be changed,
-						 * so allocate a new one.
-						 */
-						slot = MakeTupleTableSlot(tupdesc);
-						tuple = do_convert_tuple(tuple, map);
-						ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-					}
+						slot = ConvertTupleSlot(map, slot,
+												MakeTupleTableSlot(tupdesc));
 				}
 
 				insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2062,25 +2053,21 @@ 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 = inheritence_tupconv_map(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)
-				{
-					/*
-					 * Partition-specific slot's tupdesc can't be changed,
-					 * so allocate a new one.
-					 */
-					slot = MakeTupleTableSlot(tupdesc);
-					tuple = do_convert_tuple(tuple, map);
-					ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-				}
+					slot = ConvertTupleSlot(map, slot,
+											MakeTupleTableSlot(tupdesc));
 			}
 
 			insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2172,25 +2159,21 @@ 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 = inheritence_tupconv_map(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)
-						{
-							/*
-							 * Partition-specific slot's tupdesc can't be
-							 * changed, so allocate a new one.
-							 */
-							slot = MakeTupleTableSlot(tupdesc);
-							tuple = do_convert_tuple(tuple, map);
-							ExecStoreTuple(tuple, slot, InvalidBuffer, false);
-						}
+							slot = ConvertTupleSlot(map, slot,
+													MakeTupleTableSlot(tupdesc));
 					}
 
 					insertedCols = GetInsertedColumns(resultRelInfo, estate);
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index 2af7599..aa8893b 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -190,7 +190,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));
@@ -203,11 +202,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;
@@ -218,11 +216,7 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
 		 */
 		myslot = dispatch->tupslot;
 		if (myslot != NULL && map != NULL)
-		{
-			tuple = do_convert_tuple(tuple, map);
-			ExecStoreTuple(tuple, myslot, InvalidBuffer, true);
-			slot = myslot;
-		}
+			slot = ConvertTupleSlot(map, slot, myslot);
 
 		/*
 		 * Extract partition key from tuple. Expression evaluation machinery
@@ -267,16 +261,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);
-			}
 		}
 	}
 
@@ -807,36 +791,6 @@ TupConvMapForLeaf(PartitionTupleRouting *proute,
 }
 
 /*
- * 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)
-{
-	Assert(map != NULL && new_slot != NULL);
-
-	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);
-	ExecStoreTuple(tuple, new_slot, InvalidBuffer, shouldFree);
-
-	return tuple;
-}
-
-/*
  * ExecCleanupTupleRouting -- Clean up objects allocated for partition tuple
  * routing.
  *
@@ -991,9 +945,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 = inheritence_tupconv_map(RelationGetDescr(parent),
+											 tupdesc,
+											 gettext_noop("could not convert row type"));
 	}
 	else
 	{
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index e4b2d37..122ac73 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -1162,11 +1162,8 @@ lreplace:;
 			Assert(map_index >= 0 && map_index < mtstate->mt_nplans);
 			tupconv_map = tupconv_map_for_subplan(mtstate, map_index);
 			if (tupconv_map != NULL)
-				tuple = ConvertPartitionTupleSlot(tupconv_map,
-												  tuple,
-												  proute->root_tuple_slot,
-												  &slot,
-												  true);
+				slot = ConvertTupleSlot(tupconv_map->attrMap,
+										slot, proute->root_tuple_slot);
 
 			/*
 			 * Prepare for tuple routing, making it look like we're inserting
@@ -1800,7 +1797,7 @@ ExecPrepareTupleRouting(ModifyTableState *mtstate,
 		Assert(proute->partition_tuple_slots != NULL &&
 			   proute->partition_tuple_slots[partidx] != NULL);
 		new_slot = proute->partition_tuple_slots[partidx];
-		ConvertPartitionTupleSlot(map, tuple, new_slot, &slot, true);
+		slot = ConvertTupleSlot(map->attrMap, slot, new_slot);
 	}
 
 	/* Initialize information needed to handle ON CONFLICT DO UPDATE. */
diff --git a/src/include/access/tupconvert.h b/src/include/access/tupconvert.h
index 66c0ed0..77568eb 100644
--- a/src/include/access/tupconvert.h
+++ b/src/include/access/tupconvert.h
@@ -16,6 +16,7 @@
 
 #include "access/htup.h"
 #include "access/tupdesc.h"
+#include "executor/tuptable.h"
 
 
 typedef struct TupleConversionMap
@@ -38,12 +39,19 @@ extern TupleConversionMap *convert_tuples_by_name(TupleDesc indesc,
 					   TupleDesc outdesc,
 					   const char *msg);
 
+extern AttrNumber *inheritence_tupconv_map(TupleDesc indesc,
+						TupleDesc outdesc,
+						const char *msg);
+
 extern AttrNumber *convert_tuples_by_name_map(TupleDesc indesc,
 						   TupleDesc outdesc,
 						   const char *msg);
 
 extern HeapTuple do_convert_tuple(HeapTuple tuple, TupleConversionMap *map);
 
+extern TupleTableSlot *ConvertTupleSlot(AttrNumber *attrMap,
+				 TupleTableSlot *in_slot, TupleTableSlot *out_slot);
+
 extern void free_conversion_map(TupleConversionMap *map);
 
 #endif							/* TUPCONVERT_H */
diff --git a/src/include/executor/execPartition.h b/src/include/executor/execPartition.h
index c2a2dc5..f43d321 100644
--- a/src/include/executor/execPartition.h
+++ b/src/include/executor/execPartition.h
@@ -43,7 +43,7 @@ typedef struct PartitionDispatchData
 	List	   *keystate;		/* list of ExprState */
 	PartitionDesc partdesc;
 	TupleTableSlot *tupslot;
-	TupleConversionMap *tupmap;
+	AttrNumber *tupmap;
 	int		   *indexes;
 } PartitionDispatchData;
 
@@ -219,11 +219,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,

From da4eab64bacae0f783b69f4ec89ee0e11d1d1006 Mon Sep 17 00:00:00 2001
From: Amit Khandekar <amit.khandekar@enterprisedb.com>
Date: Fri, 28 Sep 2018 15:01:32 +0530
Subject: [PATCH 1/2] Allocate dedicated slots of partition tuple conversion.

Currently there is just one slot called partition_tuple_slot and
we do ExecSetSlotDescriptor every time a tuple is inserted into a
partition that requires tuple conversion.  That's excessively
inefficient during bulk inserts.

Fix that by allocating a dedicated slot for each partitions that
requires it.

Author: Amit Langote.
Reviewer: Amit Khandekar.
---
 src/backend/commands/copy.c            | 17 +++++++++----
 src/backend/executor/execMain.c        | 24 +++++++++++++++---
 src/backend/executor/execPartition.c   | 45 +++++++++++++++++++++++-----------
 src/backend/executor/nodeModifyTable.c | 27 ++++++++++++--------
 src/include/executor/execPartition.h   | 13 ++++++----
 5 files changed, 88 insertions(+), 38 deletions(-)

diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index d06662b..7fe4be5 100644
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -2690,6 +2690,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,
@@ -2864,11 +2865,17 @@ CopyFrom(CopyState cstate)
 			 * partition rowtype.  Don't free the already stored tuple as it
 			 * may still be required for a multi-insert batch.
 			 */
-			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;
+
+				Assert(proute->partition_tuple_slots != NULL &&
+					   proute->partition_tuple_slots[leaf_part_index] != NULL);
+				new_slot = proute->partition_tuple_slots[leaf_part_index];
+				tuple = ConvertPartitionTupleSlot(map, tuple, new_slot, &slot,
+												  false);
+			}
 
 			tuple->t_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
 		}
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 5443cbf..862615e 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1955,8 +1955,12 @@ ExecPartitionCheckEmitError(ResultRelInfo *resultRelInfo,
 									 gettext_noop("could not convert row type"));
 		if (map != NULL)
 		{
+			/*
+			 * Partition-specific slot's tupdesc can't be changed, so allocate
+			 * a new one.
+			 */
+			slot = MakeTupleTableSlot(tupdesc);
 			tuple = do_convert_tuple(tuple, map);
-			ExecSetSlotDescriptor(slot, tupdesc);
 			ExecStoreHeapTuple(tuple, slot, false);
 		}
 	}
@@ -2034,8 +2038,12 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 												 gettext_noop("could not convert row type"));
 					if (map != NULL)
 					{
+						/*
+						 * Partition-specific slot's tupdesc can't be changed,
+						 * so allocate a new one.
+						 */
+						slot = MakeTupleTableSlot(tupdesc);
 						tuple = do_convert_tuple(tuple, map);
-						ExecSetSlotDescriptor(slot, tupdesc);
 						ExecStoreHeapTuple(tuple, slot, false);
 					}
 				}
@@ -2082,8 +2090,12 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 											 gettext_noop("could not convert row type"));
 				if (map != NULL)
 				{
+					/*
+					 * Partition-specific slot's tupdesc can't be changed,
+					 * so allocate a new one.
+					 */
+					slot = MakeTupleTableSlot(tupdesc);
 					tuple = do_convert_tuple(tuple, map);
-					ExecSetSlotDescriptor(slot, tupdesc);
 					ExecStoreHeapTuple(tuple, slot, false);
 				}
 			}
@@ -2188,8 +2200,12 @@ ExecWithCheckOptions(WCOKind kind, ResultRelInfo *resultRelInfo,
 													 gettext_noop("could not convert row type"));
 						if (map != NULL)
 						{
+							/*
+							 * Partition-specific slot's tupdesc can't be
+							 * changed, so allocate a new one.
+							 */
+							slot = MakeTupleTableSlot(tupdesc);
 							tuple = do_convert_tuple(tuple, map);
-							ExecSetSlotDescriptor(slot, tupdesc);
 							ExecStoreHeapTuple(tuple, slot, false);
 						}
 					}
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index ec7a526..e695470 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -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)
 	{
@@ -739,6 +731,35 @@ ExecInitRoutingInfo(ModifyTableState *mtstate,
 							   gettext_noop("could not convert row type"));
 
 	/*
+	 * If partition has different rowtype than root parent, initialize a slot
+	 * dedicated to storing this partition's tuples.  The slot is used for
+	 * various operations that are applied to tuple 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.
 	 */
@@ -831,8 +852,7 @@ ConvertPartitionTupleSlot(TupleConversionMap *map,
 						  TupleTableSlot **p_my_slot,
 						  bool shouldFree)
 {
-	if (!map)
-		return tuple;
+	Assert(map != NULL && new_slot != NULL);
 
 	tuple = do_convert_tuple(tuple, map);
 
@@ -841,7 +861,6 @@ ConvertPartitionTupleSlot(TupleConversionMap *map,
 	 */
 	*p_my_slot = new_slot;
 	Assert(new_slot != NULL);
-	ExecSetSlotDescriptor(new_slot, map->outdesc);
 	ExecStoreHeapTuple(tuple, new_slot, shouldFree);
 
 	return tuple;
@@ -915,8 +934,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);
 }
 
 /*
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index bf0d5e8..9f60be6 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -1161,11 +1161,12 @@ 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)
+				tuple = ConvertPartitionTupleSlot(tupconv_map,
+												  tuple,
+												  proute->root_tuple_slot,
+												  &slot,
+												  true);
 
 			/*
 			 * Prepare for tuple routing, making it look like we're inserting
@@ -1703,6 +1704,7 @@ ExecPrepareTupleRouting(ModifyTableState *mtstate,
 	int			partidx;
 	ResultRelInfo *partrel;
 	HeapTuple	tuple;
+	TupleConversionMap *map;
 
 	/*
 	 * Determine the target partition.  If ExecFindPartition does not find a
@@ -1790,11 +1792,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];
+		ConvertPartitionTupleSlot(map, tuple, new_slot, &slot, true);
+	}
 
 	/* Initialize information needed to handle ON CONFLICT DO UPDATE. */
 	Assert(mtstate != NULL);
diff --git a/src/include/executor/execPartition.h b/src/include/executor/execPartition.h
index 89ce538..235e60f 100644
--- a/src/include/executor/execPartition.h
+++ b/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;
 
-- 
2.1.4

From 1097d60c6b94b07f5c1d28a70b545b27c373a18c Mon Sep 17 00:00:00 2001
From: Amit Khandekar <amit.khandekar@enterprisedb.com>
Date: Fri, 28 Sep 2018 15:02:57 +0530
Subject: [PATCH 2/2] Slotify partition tuple conversion.

Partition tuples are needlessly formed and deformed during tuple
conversion (do_convert_tuple), when the same operation can be done
using tuple slots. This is because the input slot might already have a
deformed tuple.

So convert directly using input and output tuple slots. Have a new
function ConvertTupleSlot() that accepts a map of AttrNumber[],
rather than TupleConversionMap.

ConvertPartitionTupleSlot() is exclusively used for conversion of
partition-to-parent and vice versa, so get rid of this.

do_convert_tuple() is used for tuples belonging to non-partition
tables as well, so leave such calls untouched. May be we can "slotify"
these tuple conversions in later commits.

Changed the PartitionDispatch->tupmap type from TupleConversionMap
to AttrNumber[] because we now don't require the input/output
descriptors since they are available in the input/output slots.

Author: Amit Khandekar.
Reviewer: Amit Langote.
---
 src/backend/access/common/tupconvert.c | 113 ++++++++++++++++++++++++++++-----
 src/backend/commands/copy.c            |  16 +++--
 src/backend/executor/execMain.c        |  87 ++++++++++---------------
 src/backend/executor/execPartition.c   |  58 ++---------------
 src/backend/executor/nodeModifyTable.c |   9 +--
 src/include/access/tupconvert.h        |   8 +++
 src/include/executor/execPartition.h   |   5 --
 7 files changed, 161 insertions(+), 135 deletions(-)

diff --git a/src/backend/access/common/tupconvert.c b/src/backend/access/common/tupconvert.c
index 3bc67b8..2100d40 100644
--- a/src/backend/access/common/tupconvert.c
+++ b/src/backend/access/common/tupconvert.c
@@ -22,6 +22,7 @@
 
 #include "access/htup_details.h"
 #include "access/tupconvert.h"
+#include "executor/tuptable.h"
 #include "utils/builtins.h"
 
 
@@ -214,6 +215,45 @@ convert_tuples_by_name(TupleDesc indesc,
 	TupleConversionMap *map;
 	AttrNumber *attrMap;
 	int			n = outdesc->natts;
+
+	/* Verify compatibility and prepare attribute-number map */
+	attrMap = inheritence_tupconv_map(indesc, outdesc, msg);
+
+	if (attrMap == NULL)
+	{
+		/* Runtime conversion is not needed */
+		return NULL;
+	}
+
+	/* Prepare the map structure */
+	map = (TupleConversionMap *) palloc(sizeof(TupleConversionMap));
+	map->indesc = indesc;
+	map->outdesc = outdesc;
+	map->attrMap = attrMap;
+	/* preallocate workspace for Datum arrays */
+	map->outvalues = (Datum *) palloc(n * sizeof(Datum));
+	map->outisnull = (bool *) palloc(n * sizeof(bool));
+	n = indesc->natts + 1;		/* +1 for NULL */
+	map->invalues = (Datum *) palloc(n * sizeof(Datum));
+	map->inisnull = (bool *) palloc(n * sizeof(bool));
+	map->invalues[0] = (Datum) 0;	/* set up the NULL entry */
+	map->inisnull[0] = true;
+
+	return map;
+}
+
+/*
+ * Returns a bare attribute map for tuple conversion. Returns NULL if
+ * conversion not required. This is a convenience routine for
+ * convert_tuples_by_name() and other functions.
+ */
+AttrNumber *
+inheritence_tupconv_map(TupleDesc indesc,
+						TupleDesc outdesc,
+						const char *msg)
+{
+	AttrNumber *attrMap;
+	int			n = outdesc->natts;
 	int			i;
 	bool		same;
 
@@ -265,22 +305,8 @@ convert_tuples_by_name(TupleDesc indesc,
 		pfree(attrMap);
 		return NULL;
 	}
-
-	/* Prepare the map structure */
-	map = (TupleConversionMap *) palloc(sizeof(TupleConversionMap));
-	map->indesc = indesc;
-	map->outdesc = outdesc;
-	map->attrMap = attrMap;
-	/* preallocate workspace for Datum arrays */
-	map->outvalues = (Datum *) palloc(n * sizeof(Datum));
-	map->outisnull = (bool *) palloc(n * sizeof(bool));
-	n = indesc->natts + 1;		/* +1 for NULL */
-	map->invalues = (Datum *) palloc(n * sizeof(Datum));
-	map->inisnull = (bool *) palloc(n * sizeof(bool));
-	map->invalues[0] = (Datum) 0;	/* set up the NULL entry */
-	map->inisnull[0] = true;
-
-	return map;
+	else
+		return attrMap;
 }
 
 /*
@@ -406,6 +432,61 @@ do_convert_tuple(HeapTuple tuple, TupleConversionMap *map)
 }
 
 /*
+ * Perform conversion of a tuple slot according to the map.
+ */
+TupleTableSlot *
+ConvertTupleSlot(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.
  */
 void
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 7fe4be5..2b370c3 100644
--- a/src/backend/commands/copy.c
+++ b/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"
@@ -2862,19 +2863,26 @@ 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.
 			 */
 			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];
-				tuple = ConvertPartitionTupleSlot(map, tuple, new_slot, &slot,
-												  false);
+				slot = ConvertTupleSlot(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);
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 862615e..bc7cb91 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1944,25 +1944,20 @@ 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 = inheritence_tupconv_map(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)
-		{
-			/*
-			 * Partition-specific slot's tupdesc can't be changed, so allocate
-			 * a new one.
-			 */
-			slot = MakeTupleTableSlot(tupdesc);
-			tuple = do_convert_tuple(tuple, map);
-			ExecStoreHeapTuple(tuple, slot, false);
-		}
+			slot = ConvertTupleSlot(map, slot, MakeTupleTableSlot(tupdesc));
 	}
 
 	insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2028,24 +2023,20 @@ 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 = inheritence_tupconv_map(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)
-					{
-						/*
-						 * Partition-specific slot's tupdesc can't be changed,
-						 * so allocate a new one.
-						 */
-						slot = MakeTupleTableSlot(tupdesc);
-						tuple = do_convert_tuple(tuple, map);
-						ExecStoreHeapTuple(tuple, slot, false);
-					}
+						slot = ConvertTupleSlot(map, slot,
+												MakeTupleTableSlot(tupdesc));
 				}
 
 				insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2079,25 +2070,21 @@ 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 = inheritence_tupconv_map(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)
-				{
-					/*
-					 * Partition-specific slot's tupdesc can't be changed,
-					 * so allocate a new one.
-					 */
-					slot = MakeTupleTableSlot(tupdesc);
-					tuple = do_convert_tuple(tuple, map);
-					ExecStoreHeapTuple(tuple, slot, false);
-				}
+					slot = ConvertTupleSlot(map, slot,
+											MakeTupleTableSlot(tupdesc));
 			}
 
 			insertedCols = GetInsertedColumns(resultRelInfo, estate);
@@ -2189,25 +2176,21 @@ 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 = inheritence_tupconv_map(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)
-						{
-							/*
-							 * Partition-specific slot's tupdesc can't be
-							 * changed, so allocate a new one.
-							 */
-							slot = MakeTupleTableSlot(tupdesc);
-							tuple = do_convert_tuple(tuple, map);
-							ExecStoreHeapTuple(tuple, slot, false);
-						}
+							slot = ConvertTupleSlot(map, slot,
+													MakeTupleTableSlot(tupdesc));
 					}
 
 					insertedCols = GetInsertedColumns(resultRelInfo, estate);
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index e695470..a90fdcf 100644
--- a/src/backend/executor/execPartition.c
+++ b/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;
 
@@ -220,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));
@@ -233,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;
@@ -248,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 = ConvertTupleSlot(map, slot, myslot);
 
 		/*
 		 * Extract partition key from tuple. Expression evaluation machinery
@@ -297,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);
-			}
 		}
 	}
 
@@ -837,36 +821,6 @@ TupConvMapForLeaf(PartitionTupleRouting *proute,
 }
 
 /*
- * 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)
-{
-	Assert(map != NULL && new_slot != NULL);
-
-	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);
-	ExecStoreHeapTuple(tuple, new_slot, shouldFree);
-
-	return tuple;
-}
-
-/*
  * ExecCleanupTupleRouting -- Clean up objects allocated for partition tuple
  * routing.
  *
@@ -1021,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 = inheritence_tupconv_map(RelationGetDescr(parent),
+											 tupdesc,
+											 gettext_noop("could not convert row type"));
 	}
 	else
 	{
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 9f60be6..d1ab9b4 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -1162,11 +1162,8 @@ lreplace:;
 			Assert(map_index >= 0 && map_index < mtstate->mt_nplans);
 			tupconv_map = tupconv_map_for_subplan(mtstate, map_index);
 			if (tupconv_map != NULL)
-				tuple = ConvertPartitionTupleSlot(tupconv_map,
-												  tuple,
-												  proute->root_tuple_slot,
-												  &slot,
-												  true);
+				slot = ConvertTupleSlot(tupconv_map->attrMap,
+										slot, proute->root_tuple_slot);
 
 			/*
 			 * Prepare for tuple routing, making it look like we're inserting
@@ -1800,7 +1797,7 @@ ExecPrepareTupleRouting(ModifyTableState *mtstate,
 		Assert(proute->partition_tuple_slots != NULL &&
 			   proute->partition_tuple_slots[partidx] != NULL);
 		new_slot = proute->partition_tuple_slots[partidx];
-		ConvertPartitionTupleSlot(map, tuple, new_slot, &slot, true);
+		slot = ConvertTupleSlot(map->attrMap, slot, new_slot);
 	}
 
 	/* Initialize information needed to handle ON CONFLICT DO UPDATE. */
diff --git a/src/include/access/tupconvert.h b/src/include/access/tupconvert.h
index 66c0ed0..77568eb 100644
--- a/src/include/access/tupconvert.h
+++ b/src/include/access/tupconvert.h
@@ -16,6 +16,7 @@
 
 #include "access/htup.h"
 #include "access/tupdesc.h"
+#include "executor/tuptable.h"
 
 
 typedef struct TupleConversionMap
@@ -38,12 +39,19 @@ extern TupleConversionMap *convert_tuples_by_name(TupleDesc indesc,
 					   TupleDesc outdesc,
 					   const char *msg);
 
+extern AttrNumber *inheritence_tupconv_map(TupleDesc indesc,
+						TupleDesc outdesc,
+						const char *msg);
+
 extern AttrNumber *convert_tuples_by_name_map(TupleDesc indesc,
 						   TupleDesc outdesc,
 						   const char *msg);
 
 extern HeapTuple do_convert_tuple(HeapTuple tuple, TupleConversionMap *map);
 
+extern TupleTableSlot *ConvertTupleSlot(AttrNumber *attrMap,
+				 TupleTableSlot *in_slot, TupleTableSlot *out_slot);
+
 extern void free_conversion_map(TupleConversionMap *map);
 
 #endif							/* TUPCONVERT_H */
diff --git a/src/include/executor/execPartition.h b/src/include/executor/execPartition.h
index 235e60f..8b4a9ca 100644
--- a/src/include/executor/execPartition.h
+++ b/src/include/executor/execPartition.h
@@ -191,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,
-- 
2.1.4

From 39591cf622ec18ee095448df3888295a3060ac1c Mon Sep 17 00:00:00 2001
From: Andres Freund <andres@anarazel.de>
Date: Tue, 2 Oct 2018 00:25:44 -0700
Subject: [PATCH] Change partition mapping code to use slots more widely and
 consistently.

Author: Amit Khandekar and Amit Langote (combined patches), 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 | 176 ++++++++++++++++++-------
 src/backend/commands/analyze.c         |   2 +-
 src/backend/commands/copy.c            |  29 +++-
 src/backend/commands/trigger.c         |   4 +-
 src/backend/executor/execExprInterp.c  |   2 +-
 src/backend/executor/execMain.c        |  79 ++++++-----
 src/backend/executor/execPartition.c   | 101 +++++---------
 src/backend/executor/nodeModifyTable.c |  24 ++--
 src/include/access/tupconvert.h        |   8 +-
 src/include/executor/execPartition.h   |  18 ++-
 src/pl/plpgsql/src/pl_exec.c           |  14 +-
 11 files changed, 270 insertions(+), 187 deletions(-)

diff --git a/src/backend/access/common/tupconvert.c b/src/backend/access/common/tupconvert.c
index 3bc67b846d4..68d977526bf 100644
--- a/src/backend/access/common/tupconvert.c
+++ b/src/backend/access/common/tupconvert.c
@@ -22,6 +22,7 @@
 
 #include "access/htup_details.h"
 #include "access/tupconvert.h"
+#include "executor/tuptable.h"
 #include "utils/builtins.h"
 
 
@@ -31,7 +32,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 +215,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);
+	attrMap = convert_tuples_by_name_map_if_req(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)
+	if (attrMap == NULL)
 	{
-		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);
+		/* runtime conversion is not needed */
 		return NULL;
 	}
 
@@ -367,11 +326,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 +431,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.
  */
diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index d11b559b20a..4fe29a3f8dc 100644
--- a/src/backend/commands/analyze.c
+++ b/src/backend/commands/analyze.c
@@ -1495,7 +1495,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;
 						}
diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index 2d5bc8add68..32706fad90f 100644
--- a/src/backend/commands/copy.c
+++ b/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);
 		}
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index b2de1cc3915..136f9f06275 100644
--- a/src/backend/commands/trigger.c
+++ b/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);
diff --git a/src/backend/executor/execExprInterp.c b/src/backend/executor/execExprInterp.c
index f597363eb10..c549e3db5d9 100644
--- a/src/backend/executor/execExprInterp.c
+++ b/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);
 	}
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 2b7cf263803..c28750e0753 100644
--- a/src/backend/executor/execMain.c
+++ b/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);
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index ec7a5267c34..832c79b41ea 100644
--- a/src/backend/executor/execPartition.c
+++ b/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
 	{
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index bf0d5e8edb5..24beb40435e 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/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);
diff --git a/src/include/access/tupconvert.h b/src/include/access/tupconvert.h
index 66c0ed0882a..34ee8e3918e 100644
--- a/src/include/access/tupconvert.h
+++ b/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);
 
diff --git a/src/include/executor/execPartition.h b/src/include/executor/execPartition.h
index 89ce53815c7..8b4a9ca0447 100644
--- a/src/include/executor/execPartition.h
+++ b/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,
diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c
index 380d1de8f4d..574234da50a 100644
--- a/src/pl/plpgsql/src/pl_exec.c
+++ b/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);
-- 
2.18.0.rc2.dirty