1 files changed, 20 insertions, 267 deletions

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);