3 files changed, 341 insertions, 267 deletions

diff --git a/src/backend/access/common/Makefile b/src/backend/access/common/Makefile
index 6c9c6f32565..fd74e14024c 100644
--- a/src/backend/access/common/Makefile
+++ b/src/backend/access/common/Makefile
@@ -13,6 +13,7 @@ top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
 OBJS = \
+	attmap.o \
 	bufmask.o \
 	detoast.o \
 	heaptuple.o \
diff --git a/src/backend/access/common/attmap.c b/src/backend/access/common/attmap.c
new file mode 100644
index 00000000000..ba909b0d4e7
--- /dev/null
+++ b/src/backend/access/common/attmap.c
@@ -0,0 +1,320 @@
+/*-------------------------------------------------------------------------
+ *
+ * attmap.c
+ *	  Attribute mapping support.
+ *
+ * This file provides utility routines to build and manage attribute
+ * mappings by comparing input and output TupleDescs.  Such mappings
+ * are typically used by DDL operating on inheritance and partition trees
+ * to do a conversion between rowtypes logically equivalent but with
+ * columns in a different order, taking into account dropped columns.
+ * They are also used by the tuple conversion routines in tupconvert.c.
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/access/common/attmap.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/attmap.h"
+#include "access/htup_details.h"
+#include "utils/builtins.h"
+
+
+static bool check_attrmap_match(TupleDesc indesc,
+								TupleDesc outdesc,
+								AttrMap *attrMap);
+
+/*
+ * make_attrmap
+ *
+ * Utility routine to allocate an attribute map in the current memory
+ * context.
+ */
+AttrMap *
+make_attrmap(int maplen)
+{
+	AttrMap    *res;
+
+	res = (AttrMap *) palloc0(sizeof(AttrMap));
+	res->maplen = maplen;
+	res->attnums = (AttrNumber *) palloc0(sizeof(AttrNumber) * maplen);
+	return res;
+}
+
+/*
+ * free_attrmap
+ *
+ * Utility routine to release an attribute map.
+ */
+void
+free_attrmap(AttrMap *map)
+{
+	pfree(map->attnums);
+	pfree(map);
+}
+
+/*
+ * build_attrmap_by_position
+ *
+ * Return a palloc'd bare attribute map for tuple conversion, matching input
+ * and output columns by position.  Dropped columns are ignored in both input
+ * and output, marked as 0.  This is normally a subroutine for
+ * convert_tuples_by_position in tupconvert.c, but it can be used standalone.
+ *
+ * Note: the errdetail messages speak of indesc as the "returned" rowtype,
+ * outdesc as the "expected" rowtype.  This is okay for current uses but
+ * might need generalization in future.
+ */
+AttrMap *
+build_attrmap_by_position(TupleDesc indesc,
+						  TupleDesc outdesc,
+						  const char *msg)
+{
+	AttrMap    *attrMap;
+	int			nincols;
+	int			noutcols;
+	int			n;
+	int			i;
+	int			j;
+	bool		same;
+
+	/*
+	 * The length is computed as the number of attributes of the expected
+	 * rowtype as it includes dropped attributes in its count.
+	 */
+	n = outdesc->natts;
+	attrMap = make_attrmap(n);
+
+	j = 0;						/* j is next physical input attribute */
+	nincols = noutcols = 0;		/* these count non-dropped attributes */
+	same = true;
+	for (i = 0; i < n; i++)
+	{
+		Form_pg_attribute att = TupleDescAttr(outdesc, i);
+		Oid			atttypid;
+		int32		atttypmod;
+
+		if (att->attisdropped)
+			continue;			/* attrMap->attnums[i] is already 0 */
+		noutcols++;
+		atttypid = att->atttypid;
+		atttypmod = att->atttypmod;
+		for (; j < indesc->natts; j++)
+		{
+			att = TupleDescAttr(indesc, j);
+			if (att->attisdropped)
+				continue;
+			nincols++;
+
+			/* Found matching column, now check type */
+			if (atttypid != att->atttypid ||
+				(atttypmod != att->atttypmod && atttypmod >= 0))
+				ereport(ERROR,
+						(errcode(ERRCODE_DATATYPE_MISMATCH),
+						 errmsg_internal("%s", _(msg)),
+						 errdetail("Returned type %s does not match expected type %s in column %d.",
+								   format_type_with_typemod(att->atttypid,
+															att->atttypmod),
+								   format_type_with_typemod(atttypid,
+															atttypmod),
+								   noutcols)));
+			attrMap->attnums[i] = (AttrNumber) (j + 1);
+			j++;
+			break;
+		}
+		if (attrMap->attnums[i] == 0)
+			same = false;		/* we'll complain below */
+	}
+
+	/* Check for unused input columns */
+	for (; j < indesc->natts; j++)
+	{
+		if (TupleDescAttr(indesc, j)->attisdropped)
+			continue;
+		nincols++;
+		same = false;			/* we'll complain below */
+	}
+
+	/* Report column count mismatch using the non-dropped-column counts */
+	if (!same)
+		ereport(ERROR,
+				(errcode(ERRCODE_DATATYPE_MISMATCH),
+				 errmsg_internal("%s", _(msg)),
+				 errdetail("Number of returned columns (%d) does not match "
+						   "expected column count (%d).",
+						   nincols, noutcols)));
+
+	/* Check if the map has a one-to-one match */
+	if (check_attrmap_match(indesc, outdesc, attrMap))
+	{
+		/* Runtime conversion is not needed */
+		free_attrmap(attrMap);
+		return NULL;
+	}
+
+	return attrMap;
+}
+
+/*
+ * build_attrmap_by_name
+ *
+ * Return a palloc'd bare attribute map for tuple conversion, matching input
+ * and output columns by name.  (Dropped columns are ignored in both input and
+ * output.)  This is normally a subroutine for convert_tuples_by_name in
+ * tupconvert.c, but can be used standalone.
+ */
+AttrMap *
+build_attrmap_by_name(TupleDesc indesc,
+					  TupleDesc outdesc)
+{
+	AttrMap    *attrMap;
+	int			outnatts;
+	int			innatts;
+	int			i;
+	int			nextindesc = -1;
+
+	outnatts = outdesc->natts;
+	innatts = indesc->natts;
+
+	attrMap = make_attrmap(outnatts);
+	for (i = 0; i < outnatts; i++)
+	{
+		Form_pg_attribute outatt = TupleDescAttr(outdesc, i);
+		char	   *attname;
+		Oid			atttypid;
+		int32		atttypmod;
+		int			j;
+
+		if (outatt->attisdropped)
+			continue;			/* attrMap->attnums[i] is already 0 */
+		attname = NameStr(outatt->attname);
+		atttypid = outatt->atttypid;
+		atttypmod = outatt->atttypmod;
+
+		/*
+		 * Now search for an attribute with the same name in the indesc. It
+		 * seems likely that a partitioned table will have the attributes in
+		 * the same order as the partition, so the search below is optimized
+		 * for that case.  It is possible that columns are dropped in one of
+		 * the relations, but not the other, so we use the 'nextindesc'
+		 * counter to track the starting point of the search.  If the inner
+		 * loop encounters dropped columns then it will have to skip over
+		 * them, but it should leave 'nextindesc' at the correct position for
+		 * the next outer loop.
+		 */
+		for (j = 0; j < innatts; j++)
+		{
+			Form_pg_attribute inatt;
+
+			nextindesc++;
+			if (nextindesc >= innatts)
+				nextindesc = 0;
+
+			inatt = TupleDescAttr(indesc, nextindesc);
+			if (inatt->attisdropped)
+				continue;
+			if (strcmp(attname, NameStr(inatt->attname)) == 0)
+			{
+				/* Found it, check type */
+				if (atttypid != inatt->atttypid || atttypmod != inatt->atttypmod)
+					ereport(ERROR,
+							(errcode(ERRCODE_DATATYPE_MISMATCH),
+							 errmsg("could not convert row type"),
+							 errdetail("Attribute \"%s\" of type %s does not match corresponding attribute of type %s.",
+									   attname,
+									   format_type_be(outdesc->tdtypeid),
+									   format_type_be(indesc->tdtypeid))));
+				attrMap->attnums[i] = inatt->attnum;
+				break;
+			}
+		}
+		if (attrMap->attnums[i] == 0)
+			ereport(ERROR,
+					(errcode(ERRCODE_DATATYPE_MISMATCH),
+					 errmsg("could not convert row type"),
+					 errdetail("Attribute \"%s\" of type %s does not exist in type %s.",
+							   attname,
+							   format_type_be(outdesc->tdtypeid),
+							   format_type_be(indesc->tdtypeid))));
+	}
+	return attrMap;
+}
+
+/*
+ * build_attrmap_by_name_if_req
+ *
+ * Returns mapping created by build_attrmap_by_name, or NULL if no
+ * conversion is required.  This is a convenience routine for
+ * convert_tuples_by_name() in tupconvert.c and other functions, but it
+ * can be used standalone.
+ */
+AttrMap *
+build_attrmap_by_name_if_req(TupleDesc indesc,
+							 TupleDesc outdesc)
+{
+	AttrMap    *attrMap;
+
+	/* Verify compatibility and prepare attribute-number map */
+	attrMap = build_attrmap_by_name(indesc, outdesc);
+
+	/* Check if the map has a one-to-one match */
+	if (check_attrmap_match(indesc, outdesc, attrMap))
+	{
+		/* Runtime conversion is not needed */
+		free_attrmap(attrMap);
+		return NULL;
+	}
+
+	return attrMap;
+}
+
+/*
+ * check_attrmap_match
+ *
+ * Check to see if the map is a one-to-one match, in which case we need
+ * not to do a tuple conversion, and the attribute map is not necessary.
+ */
+static bool
+check_attrmap_match(TupleDesc indesc,
+					TupleDesc outdesc,
+					AttrMap *attrMap)
+{
+	int			i;
+
+	/* no match if attribute numbers are not the same */
+	if (indesc->natts != outdesc->natts)
+		return false;
+
+	for (i = 0; i < attrMap->maplen; i++)
+	{
+		Form_pg_attribute inatt;
+		Form_pg_attribute outatt;
+
+		if (attrMap->attnums[i] == (i + 1))
+			continue;
+
+		/*
+		 * If it's a dropped column and the corresponding input column is also
+		 * dropped, we don't need a conversion.  However, attlen and attalign
+		 * must agree.
+		 */
+		inatt = TupleDescAttr(indesc, i);
+		outatt = TupleDescAttr(outdesc, i);
+		if (attrMap->attnums[i] == 0 &&
+			inatt->attisdropped &&
+			inatt->attlen == outatt->attlen &&
+			inatt->attalign == outatt->attalign)
+			continue;
+
+		return false;
+	}
+
+	return true;
+}
diff --git a/src/backend/access/common/tupconvert.c b/src/backend/access/common/tupconvert.c
index 0ec9cd5870d..85b194f3944 100644
--- a/src/backend/access/common/tupconvert.c
+++ b/src/backend/access/common/tupconvert.c
@@ -18,20 +18,18 @@
  */
 #include "postgres.h"
 
-#include "access/htup_details.h"
 #include "access/tupconvert.h"
 #include "executor/tuptable.h"
-#include "utils/builtins.h"
 
 
 /*
  * The conversion setup routines have the following common API:
  *
- * 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 execute_attr_map_tuple
- * to perform the conversion.
+ * The setup routine checks using attmap.c 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
+ * execute_attr_map_tuple or execute_attr_map_slot to perform the conversion.
  *
  * The TupleConversionMap, if needed, is palloc'd in the caller's memory
  * context.  Also, the given tuple descriptors are referenced by the map,
@@ -56,10 +54,6 @@
 /*
  * Set up for tuple conversion, matching input and output columns by
  * position.  (Dropped columns are ignored in both input and output.)
- *
- * Note: the errdetail messages speak of indesc as the "returned" rowtype,
- * outdesc as the "expected" rowtype.  This is okay for current uses but
- * might need generalization in future.
  */
 TupleConversionMap *
 convert_tuples_by_position(TupleDesc indesc,
@@ -67,113 +61,15 @@ convert_tuples_by_position(TupleDesc indesc,
 						   const char *msg)
 {
 	TupleConversionMap *map;
-	AttrNumber *attrMap;
-	int			nincols;
-	int			noutcols;
 	int			n;
-	int			i;
-	int			j;
-	bool		same;
+	AttrMap    *attrMap;
 
 	/* Verify compatibility and prepare attribute-number map */
-	n = outdesc->natts;
-	attrMap = (AttrNumber *) palloc0(n * sizeof(AttrNumber));
-	j = 0;						/* j is next physical input attribute */
-	nincols = noutcols = 0;		/* these count non-dropped attributes */
-	same = true;
-	for (i = 0; i < n; i++)
-	{
-		Form_pg_attribute att = TupleDescAttr(outdesc, i);
-		Oid			atttypid;
-		int32		atttypmod;
-
-		if (att->attisdropped)
-			continue;			/* attrMap[i] is already 0 */
-		noutcols++;
-		atttypid = att->atttypid;
-		atttypmod = att->atttypmod;
-		for (; j < indesc->natts; j++)
-		{
-			att = TupleDescAttr(indesc, j);
-			if (att->attisdropped)
-				continue;
-			nincols++;
-			/* Found matching column, check type */
-			if (atttypid != att->atttypid ||
-				(atttypmod != att->atttypmod && atttypmod >= 0))
-				ereport(ERROR,
-						(errcode(ERRCODE_DATATYPE_MISMATCH),
-						 errmsg_internal("%s", _(msg)),
-						 errdetail("Returned type %s does not match expected type %s in column %d.",
-								   format_type_with_typemod(att->atttypid,
-															att->atttypmod),
-								   format_type_with_typemod(atttypid,
-															atttypmod),
-								   noutcols)));
-			attrMap[i] = (AttrNumber) (j + 1);
-			j++;
-			break;
-		}
-		if (attrMap[i] == 0)
-			same = false;		/* we'll complain below */
-	}
-
-	/* Check for unused input columns */
-	for (; j < indesc->natts; j++)
-	{
-		if (TupleDescAttr(indesc, j)->attisdropped)
-			continue;
-		nincols++;
-		same = false;			/* we'll complain below */
-	}
-
-	/* Report column count mismatch using the non-dropped-column counts */
-	if (!same)
-		ereport(ERROR,
-				(errcode(ERRCODE_DATATYPE_MISMATCH),
-				 errmsg_internal("%s", _(msg)),
-				 errdetail("Number of returned columns (%d) does not match "
-						   "expected column count (%d).",
-						   nincols, noutcols)));
-
-	/*
-	 * Check to see if the map is one-to-one, in which case we need not do a
-	 * tuple conversion.
-	 */
-	if (indesc->natts == outdesc->natts)
-	{
-		for (i = 0; i < n; i++)
-		{
-			Form_pg_attribute inatt;
-			Form_pg_attribute outatt;
-
-			if (attrMap[i] == (i + 1))
-				continue;
+	attrMap = build_attrmap_by_position(indesc, outdesc, msg);
 
-			/*
-			 * 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)
+	if (attrMap == NULL)
 	{
-		/* Runtime conversion is not needed */
-		pfree(attrMap);
+		/* runtime conversion is not needed */
 		return NULL;
 	}
 
@@ -183,6 +79,7 @@ convert_tuples_by_position(TupleDesc indesc,
 	map->outdesc = outdesc;
 	map->attrMap = attrMap;
 	/* preallocate workspace for Datum arrays */
+	n = outdesc->natts + 1;		/* +1 for NULL */
 	map->outvalues = (Datum *) palloc(n * sizeof(Datum));
 	map->outisnull = (bool *) palloc(n * sizeof(bool));
 	n = indesc->natts + 1;		/* +1 for NULL */
@@ -206,11 +103,11 @@ convert_tuples_by_name(TupleDesc indesc,
 					   TupleDesc outdesc)
 {
 	TupleConversionMap *map;
-	AttrNumber *attrMap;
+	AttrMap    *attrMap;
 	int			n = outdesc->natts;
 
 	/* Verify compatibility and prepare attribute-number map */
-	attrMap = convert_tuples_by_name_map_if_req(indesc, outdesc);
+	attrMap = build_attrmap_by_name_if_req(indesc, outdesc);
 
 	if (attrMap == NULL)
 	{
@@ -236,157 +133,12 @@ convert_tuples_by_name(TupleDesc indesc,
 }
 
 /*
- * Return a palloc'd bare attribute map for tuple conversion, matching input
- * and output columns by name.  (Dropped columns are ignored in both input and
- * output.)  This is normally a subroutine for convert_tuples_by_name, but can
- * be used standalone.
- */
-AttrNumber *
-convert_tuples_by_name_map(TupleDesc indesc,
-						   TupleDesc outdesc)
-{
-	AttrNumber *attrMap;
-	int			outnatts;
-	int			innatts;
-	int			i;
-	int			nextindesc = -1;
-
-	outnatts = outdesc->natts;
-	innatts = indesc->natts;
-
-	attrMap = (AttrNumber *) palloc0(outnatts * sizeof(AttrNumber));
-	for (i = 0; i < outnatts; i++)
-	{
-		Form_pg_attribute outatt = TupleDescAttr(outdesc, i);
-		char	   *attname;
-		Oid			atttypid;
-		int32		atttypmod;
-		int			j;
-
-		if (outatt->attisdropped)
-			continue;			/* attrMap[i] is already 0 */
-		attname = NameStr(outatt->attname);
-		atttypid = outatt->atttypid;
-		atttypmod = outatt->atttypmod;
-
-		/*
-		 * Now search for an attribute with the same name in the indesc. It
-		 * seems likely that a partitioned table will have the attributes in
-		 * the same order as the partition, so the search below is optimized
-		 * for that case.  It is possible that columns are dropped in one of
-		 * the relations, but not the other, so we use the 'nextindesc'
-		 * counter to track the starting point of the search.  If the inner
-		 * loop encounters dropped columns then it will have to skip over
-		 * them, but it should leave 'nextindesc' at the correct position for
-		 * the next outer loop.
-		 */
-		for (j = 0; j < innatts; j++)
-		{
-			Form_pg_attribute inatt;
-
-			nextindesc++;
-			if (nextindesc >= innatts)
-				nextindesc = 0;
-
-			inatt = TupleDescAttr(indesc, nextindesc);
-			if (inatt->attisdropped)
-				continue;
-			if (strcmp(attname, NameStr(inatt->attname)) == 0)
-			{
-				/* Found it, check type */
-				if (atttypid != inatt->atttypid || atttypmod != inatt->atttypmod)
-					ereport(ERROR,
-							(errcode(ERRCODE_DATATYPE_MISMATCH),
-							 errmsg("could not convert row type"),
-							 errdetail("Attribute \"%s\" of type %s does not match corresponding attribute of type %s.",
-									   attname,
-									   format_type_be(outdesc->tdtypeid),
-									   format_type_be(indesc->tdtypeid))));
-				attrMap[i] = inatt->attnum;
-				break;
-			}
-		}
-		if (attrMap[i] == 0)
-			ereport(ERROR,
-					(errcode(ERRCODE_DATATYPE_MISMATCH),
-					 errmsg("could not convert row type"),
-					 errdetail("Attribute \"%s\" of type %s does not exist in type %s.",
-							   attname,
-							   format_type_be(outdesc->tdtypeid),
-							   format_type_be(indesc->tdtypeid))));
-	}
-	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)
-{
-	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);
-
-	/*
-	 * Check to see if the map is one-to-one, in which case we need not do a
-	 * tuple conversion.
-	 */
-	if (indesc->natts == outdesc->natts)
-	{
-		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
 execute_attr_map_tuple(HeapTuple tuple, TupleConversionMap *map)
 {
-	AttrNumber *attrMap = map->attrMap;
+	AttrMap    *attrMap = map->attrMap;
 	Datum	   *invalues = map->invalues;
 	bool	   *inisnull = map->inisnull;
 	Datum	   *outvalues = map->outvalues;
@@ -404,9 +156,10 @@ execute_attr_map_tuple(HeapTuple tuple, TupleConversionMap *map)
 	/*
 	 * Transpose into proper fields of the new tuple.
 	 */
-	for (i = 0; i < outnatts; i++)
+	Assert(attrMap->maplen == outnatts);
+	for (i = 0; i < attrMap->maplen; i++)
 	{
-		int			j = attrMap[i];
+		int			j = attrMap->attnums[i];
 
 		outvalues[i] = invalues[j];
 		outisnull[i] = inisnull[j];
@@ -422,7 +175,7 @@ execute_attr_map_tuple(HeapTuple tuple, TupleConversionMap *map)
  * Perform conversion of a tuple slot according to the map.
  */
 TupleTableSlot *
-execute_attr_map_slot(AttrNumber *attrMap,
+execute_attr_map_slot(AttrMap *attrMap,
 					  TupleTableSlot *in_slot,
 					  TupleTableSlot *out_slot)
 {
@@ -454,9 +207,9 @@ execute_attr_map_slot(AttrNumber *attrMap,
 	/* Transpose into proper fields of the out slot. */
 	for (i = 0; i < outnatts; i++)
 	{
-		int			j = attrMap[i] - 1;
+		int			j = attrMap->attnums[i] - 1;
 
-		/* attrMap[i] == 0 means it's a NULL datum. */
+		/* attrMap->attnums[i] == 0 means it's a NULL datum. */
 		if (j == -1)
 		{
 			outvalues[i] = (Datum) 0;
@@ -481,7 +234,7 @@ void
 free_conversion_map(TupleConversionMap *map)
 {
 	/* indesc and outdesc are not ours to free */
-	pfree(map->attrMap);
+	free_attrmap(map->attrMap);
 	pfree(map->invalues);
 	pfree(map->inisnull);
 	pfree(map->outvalues);