Rework manipulation and structure of attribute mappings

From fc874cc3f3477474cc6db3ad9367d9c015a9a168 Mon Sep 17 00:00:00 2001
From: amit <amitlangote09@gmail.com>
Date: Mon, 25 Nov 2019 16:38:32 +0900
Subject: [PATCH 2/2] Move more code to attmap.c

---
 src/backend/access/common/attmap.c     | 287 ++++++++++++++++++++++++++++++++-
 src/backend/access/common/tupconvert.c | 274 ++-----------------------------
 src/backend/catalog/partition.c        |   2 +-
 src/backend/catalog/pg_constraint.c    |   1 -
 src/backend/commands/tablecmds.c       |   2 +-
 src/backend/jit/llvm/llvmjit_expr.c    |   1 -
 src/include/access/attmap.h            |  12 +-
 src/include/access/tupconvert.h        |   5 -
 8 files changed, 305 insertions(+), 279 deletions(-)

diff --git a/src/backend/access/common/attmap.c b/src/backend/access/common/attmap.c
index 06df1dcfa4..9f47a97662 100644
--- a/src/backend/access/common/attmap.c
+++ b/src/backend/access/common/attmap.c
@@ -3,10 +3,11 @@
  * attmap.c
  *	  Attribute mapping support.
  *
- * This file provides utility routines to build and managed attribute
- * mappings which are used when working on inheritance and partition trees
- * for cloned object creation or tuple attribute comparison for runtime
- * operations.
+ * 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 convert fully
+ * transformed expression trees from parent rowtype to child rowtype or vice
+ * versa.  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
@@ -20,6 +21,284 @@
 #include "postgres.h"
 
 #include "access/attmap.h"
+#include "access/htup_details.h"
+#include "utils/builtins.h"
+
+/*
+ * 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.)  This is normally a subroutine for convert_tuples_by_position
+ * in tupconvert.c, but 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 *
+convert_tuples_by_position_map(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, 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 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->attnums[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->attnums[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 */
+		free_attrmap(attrMap);
+		return NULL;
+	}
+
+	return attrMap;
+}
+
+/*
+ * 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 *
+convert_tuples_by_name_map(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;
+}
+
+/*
+ * 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() in tupconvert.c and other functions.
+ */
+AttrMap *
+convert_tuples_by_name_map_if_req(TupleDesc indesc,
+								  TupleDesc outdesc)
+{
+	AttrMap	   *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->attnums[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->attnums[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 */
+		free_attrmap(attrMap);
+		return NULL;
+	}
+	else
+		return attrMap;
+}
 
 /*
  * make_attrmap
diff --git a/src/backend/access/common/tupconvert.c b/src/backend/access/common/tupconvert.c
index 94b90d6115..824bfabd94 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,119 +61,15 @@ convert_tuples_by_position(TupleDesc indesc,
 						   const char *msg)
 {
 	TupleConversionMap *map;
-	AttrMap    *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 = convert_tuples_by_position_map(indesc, outdesc, msg);
 
-	/*
-	 * The length is computed as the number of attributes of the expected
-	 * rowtype as it includes dropped attributes in its count.
-	 */
-	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++)
+	if (attrMap == NULL)
 	{
-		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, 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 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->attnums[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->attnums[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 */
-		free_attrmap(attrMap);
+		/* runtime conversion is not needed */
 		return NULL;
 	}
 
@@ -189,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 */
@@ -242,151 +133,6 @@ 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.
- */
-AttrMap *
-convert_tuples_by_name_map(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;
-}
-
-/*
- * 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.
- */
-AttrMap *
-convert_tuples_by_name_map_if_req(TupleDesc indesc,
-								  TupleDesc outdesc)
-{
-	AttrMap	   *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->attnums[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->attnums[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 */
-		free_attrmap(attrMap);
-		return NULL;
-	}
-	else
-		return attrMap;
-}
-
-/*
  * Perform conversion of a tuple according to the map.
  */
 HeapTuple
diff --git a/src/backend/catalog/partition.c b/src/backend/catalog/partition.c
index 2b14e9f31a..7fb545bd2f 100644
--- a/src/backend/catalog/partition.c
+++ b/src/backend/catalog/partition.c
@@ -14,11 +14,11 @@
 */
 #include "postgres.h"
 
+#include "access/attmap.h"
 #include "access/genam.h"
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
-#include "access/tupconvert.h"
 #include "catalog/indexing.h"
 #include "catalog/partition.h"
 #include "catalog/pg_inherits.h"
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 56568b0105..25c52134f0 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -18,7 +18,6 @@
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
-#include "access/tupconvert.h"
 #include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 0f1b1a7cd3..6ba353d48c 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -14,6 +14,7 @@
  */
 #include "postgres.h"
 
+#include "access/attmap.h"
 #include "access/genam.h"
 #include "access/heapam.h"
 #include "access/heapam_xlog.h"
@@ -22,7 +23,6 @@
 #include "access/relscan.h"
 #include "access/sysattr.h"
 #include "access/tableam.h"
-#include "access/tupconvert.h"
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "catalog/catalog.h"
diff --git a/src/backend/jit/llvm/llvmjit_expr.c b/src/backend/jit/llvm/llvmjit_expr.c
index a9d362100a..ffd887c71a 100644
--- a/src/backend/jit/llvm/llvmjit_expr.c
+++ b/src/backend/jit/llvm/llvmjit_expr.c
@@ -20,7 +20,6 @@
 
 #include "access/htup_details.h"
 #include "access/nbtree.h"
-#include "access/tupconvert.h"
 #include "catalog/objectaccess.h"
 #include "catalog/pg_type.h"
 #include "executor/execExpr.h"
diff --git a/src/include/access/attmap.h b/src/include/access/attmap.h
index 8e4882f096..d103d7b6eb 100644
--- a/src/include/access/attmap.h
+++ b/src/include/access/attmap.h
@@ -1,6 +1,6 @@
 /*-------------------------------------------------------------------------
  *
- * attnum.h
+ * attmap.h
  *	  Definitions for PostgreSQL attribute mappings
  *
  *
@@ -15,6 +15,7 @@
 #define ATTMAP_H
 
 #include "access/attnum.h"
+#include "access/tupdesc.h"
 
 /*
  * Attribute mapping structure
@@ -35,7 +36,14 @@ typedef struct AttrMap
 	int				maplen;
 } AttrMap;
 
-/* utility routines to manipulate attribute maps */
+extern AttrMap *convert_tuples_by_name_map(TupleDesc indesc,
+										   TupleDesc outdesc);
+extern AttrMap *convert_tuples_by_name_map_if_req(TupleDesc indesc,
+												  TupleDesc outdesc);
+extern AttrMap *convert_tuples_by_position_map(TupleDesc indesc,
+							   TupleDesc outdesc,
+							   const char *msg);
+
 extern AttrMap *make_attrmap(int maplen);
 extern void		free_attrmap(AttrMap *map);
 
diff --git a/src/include/access/tupconvert.h b/src/include/access/tupconvert.h
index f48cba967d..5ed7fe6d39 100644
--- a/src/include/access/tupconvert.h
+++ b/src/include/access/tupconvert.h
@@ -39,11 +39,6 @@ extern TupleConversionMap *convert_tuples_by_position(TupleDesc indesc,
 extern TupleConversionMap *convert_tuples_by_name(TupleDesc indesc,
 												  TupleDesc outdesc);
 
-extern AttrMap *convert_tuples_by_name_map(TupleDesc indesc,
-										   TupleDesc outdesc);
-extern AttrMap *convert_tuples_by_name_map_if_req(TupleDesc indesc,
-												  TupleDesc outdesc);
-
 extern HeapTuple execute_attr_map_tuple(HeapTuple tuple, TupleConversionMap *map);
 extern TupleTableSlot *execute_attr_map_slot(AttrMap *attrMap,
 											 TupleTableSlot *in_slot,
-- 
2.11.0

From 9b7414a8df8da37147370541d92c4c18f9eec7ac Mon Sep 17 00:00:00 2001
From: Michael Paquier <michael@paquier.xyz>
Date: Wed, 4 Dec 2019 17:00:26 +0900
Subject: [PATCH v2] Rework attribute mappings

---
 src/backend/access/common/Makefile         |   1 +
 src/backend/access/common/attmap.c         | 320 +++++++++++++++++++++
 src/backend/access/common/tupconvert.c     | 287 ++----------------
 src/backend/catalog/index.c                |  12 +-
 src/backend/catalog/partition.c            |  11 +-
 src/backend/catalog/pg_constraint.c        |   1 -
 src/backend/commands/indexcmds.c           |  16 +-
 src/backend/commands/tablecmds.c           | 100 ++++---
 src/backend/executor/execMain.c            |  22 +-
 src/backend/executor/execPartition.c       |  57 ++--
 src/backend/jit/llvm/llvmjit_expr.c        |   1 -
 src/backend/parser/parse_utilcmd.c         |  18 +-
 src/backend/replication/logical/relation.c |  10 +-
 src/backend/replication/logical/worker.c   |   9 +-
 src/backend/rewrite/rewriteManip.c         |  12 +-
 src/include/access/attmap.h                |  51 ++++
 src/include/access/tupconvert.h            |  13 +-
 src/include/catalog/index.h                |   2 +-
 src/include/parser/parse_utilcmd.h         |   3 +-
 src/include/replication/logicalrelation.h  |   2 +-
 src/include/rewrite/rewriteManip.h         |   3 +-
 21 files changed, 531 insertions(+), 420 deletions(-)
 create mode 100644 src/backend/access/common/attmap.c
 create mode 100644 src/include/access/attmap.h

diff --git a/src/backend/access/common/Makefile b/src/backend/access/common/Makefile
index 6c9c6f3256..fd74e14024 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 0000000000..33ffef9039
--- /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 convert fully
+ * transformed expression trees from parent rowtype to child rowtype or
+ * vice-versa.  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 convertion.  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 0ec9cd5870..85b194f394 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)
 	{
@@ -235,158 +132,13 @@ convert_tuples_by_name(TupleDesc indesc,
 	return map;
 }
 
-/*
- * 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);
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index e9955707fa..78b45ae6e1 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -2382,13 +2382,13 @@ BuildDummyIndexInfo(Relation index)
  * Note: passing collations and opfamilies separately is a kludge.  Adding
  * them to IndexInfo may result in better coding here and elsewhere.
  *
- * Use convert_tuples_by_name_map(index2, index1) to build the attmap.
+ * Use build_attrmap_by_name(index2, index1) to build the attmap.
  */
 bool
 CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
 				 Oid *collations1, Oid *collations2,
 				 Oid *opfamilies1, Oid *opfamilies2,
-				 AttrNumber *attmap, int maplen)
+				 AttrMap *attmap)
 {
 	int			i;
 
@@ -2415,12 +2415,12 @@ CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
 	 */
 	for (i = 0; i < info1->ii_NumIndexAttrs; i++)
 	{
-		if (maplen < info2->ii_IndexAttrNumbers[i])
+		if (attmap->maplen < info2->ii_IndexAttrNumbers[i])
 			elog(ERROR, "incorrect attribute map");
 
 		/* ignore expressions at this stage */
 		if ((info1->ii_IndexAttrNumbers[i] != InvalidAttrNumber) &&
-			(attmap[info2->ii_IndexAttrNumbers[i] - 1] !=
+			(attmap->attnums[info2->ii_IndexAttrNumbers[i] - 1] !=
 			 info1->ii_IndexAttrNumbers[i]))
 			return false;
 
@@ -2446,7 +2446,7 @@ CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
 		Node	   *mapped;
 
 		mapped = map_variable_attnos((Node *) info2->ii_Expressions,
-									 1, 0, attmap, maplen,
+									 1, 0, attmap,
 									 InvalidOid, &found_whole_row);
 		if (found_whole_row)
 		{
@@ -2470,7 +2470,7 @@ CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
 		Node	   *mapped;
 
 		mapped = map_variable_attnos((Node *) info2->ii_Predicate,
-									 1, 0, attmap, maplen,
+									 1, 0, attmap,
 									 InvalidOid, &found_whole_row);
 		if (found_whole_row)
 		{
diff --git a/src/backend/catalog/partition.c b/src/backend/catalog/partition.c
index 5dfa4499fd..7657608dd7 100644
--- a/src/backend/catalog/partition.c
+++ b/src/backend/catalog/partition.c
@@ -14,11 +14,11 @@
 */
 #include "postgres.h"
 
+#include "access/attmap.h"
 #include "access/genam.h"
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
-#include "access/tupconvert.h"
 #include "catalog/indexing.h"
 #include "catalog/partition.h"
 #include "catalog/pg_inherits.h"
@@ -206,14 +206,13 @@ map_partition_varattnos(List *expr, int fromrel_varno,
 
 	if (expr != NIL)
 	{
-		AttrNumber *part_attnos;
+		AttrMap    *part_attmap;
 
-		part_attnos = convert_tuples_by_name_map(RelationGetDescr(to_rel),
-												 RelationGetDescr(from_rel));
+		part_attmap = build_attrmap_by_name(RelationGetDescr(to_rel),
+											RelationGetDescr(from_rel));
 		expr = (List *) map_variable_attnos((Node *) expr,
 											fromrel_varno, 0,
-											part_attnos,
-											RelationGetDescr(from_rel)->natts,
+											part_attmap,
 											RelationGetForm(to_rel)->reltype,
 											&my_found_whole_row);
 	}
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 56568b0105..25c52134f0 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -18,7 +18,6 @@
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
-#include "access/tupconvert.h"
 #include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 374e2d0efe..8f242aef1e 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -1060,9 +1060,8 @@ DefineIndex(Oid relationId,
 				Relation	childrel;
 				List	   *childidxs;
 				ListCell   *cell;
-				AttrNumber *attmap;
+				AttrMap    *attmap;
 				bool		found = false;
-				int			maplen;
 
 				childrel = table_open(childRelid, lockmode);
 
@@ -1087,9 +1086,8 @@ DefineIndex(Oid relationId,
 
 				childidxs = RelationGetIndexList(childrel);
 				attmap =
-					convert_tuples_by_name_map(RelationGetDescr(childrel),
-											   parentDesc);
-				maplen = parentDesc->natts;
+					build_attrmap_by_name(RelationGetDescr(childrel),
+										  parentDesc);
 
 				foreach(cell, childidxs)
 				{
@@ -1108,7 +1106,7 @@ DefineIndex(Oid relationId,
 										 collationObjectId,
 										 cldidx->rd_opfamily,
 										 opfamOids,
-										 attmap, maplen))
+										 attmap))
 					{
 						Oid			cldConstrOid = InvalidOid;
 
@@ -1193,7 +1191,7 @@ DefineIndex(Oid relationId,
 						{
 							ielem->expr =
 								map_variable_attnos((Node *) ielem->expr,
-													1, 0, attmap, maplen,
+													1, 0, attmap,
 													InvalidOid,
 													&found_whole_row);
 							if (found_whole_row)
@@ -1202,7 +1200,7 @@ DefineIndex(Oid relationId,
 					}
 					childStmt->whereClause =
 						map_variable_attnos(stmt->whereClause, 1, 0,
-											attmap, maplen,
+											attmap,
 											InvalidOid, &found_whole_row);
 					if (found_whole_row)
 						elog(ERROR, "cannot convert whole-row table reference");
@@ -1217,7 +1215,7 @@ DefineIndex(Oid relationId,
 
 				pgstat_progress_update_param(PROGRESS_CREATEIDX_PARTITIONS_DONE,
 											 i + 1);
-				pfree(attmap);
+				free_attrmap(attmap);
 			}
 
 			/*
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 5440eb9015..6b23edfbc7 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -14,6 +14,7 @@
  */
 #include "postgres.h"
 
+#include "access/attmap.h"
 #include "access/genam.h"
 #include "access/heapam.h"
 #include "access/heapam_xlog.h"
@@ -22,7 +23,6 @@
 #include "access/relscan.h"
 #include "access/sysattr.h"
 #include "access/tableam.h"
-#include "access/tupconvert.h"
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "catalog/catalog.h"
@@ -1072,7 +1072,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		foreach(cell, idxlist)
 		{
 			Relation	idxRel = index_open(lfirst_oid(cell), AccessShareLock);
-			AttrNumber *attmap;
+			AttrMap    *attmap;
 			IndexStmt  *idxstmt;
 			Oid			constraintOid;
 
@@ -1092,12 +1092,11 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 				}
 			}
 
-			attmap = convert_tuples_by_name_map(RelationGetDescr(rel),
-												RelationGetDescr(parent));
+			attmap = build_attrmap_by_name(RelationGetDescr(rel),
+										   RelationGetDescr(parent));
 			idxstmt =
 				generateClonedIndexStmt(NULL, idxRel,
-										attmap, RelationGetDescr(parent)->natts,
-										&constraintOid);
+										attmap, &constraintOid);
 			DefineIndex(RelationGetRelid(rel),
 						idxstmt,
 						InvalidOid,
@@ -2158,7 +2157,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 		Relation	relation;
 		TupleDesc	tupleDesc;
 		TupleConstr *constr;
-		AttrNumber *newattno;
+		AttrMap    *newattmap;
 		AttrNumber	parent_attno;
 
 		/* caller already got lock */
@@ -2239,12 +2238,11 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 		constr = tupleDesc->constr;
 
 		/*
-		 * newattno[] will contain the child-table attribute numbers for the
-		 * attributes of this parent table.  (They are not the same for
-		 * parents after the first one, nor if we have dropped columns.)
+		 * newattmap->attnums[] will contain the child-table attribute numbers
+		 * for the attributes of this parent table.  (They are not the same
+		 * for parents after the first one, nor if we have dropped columns.)
 		 */
-		newattno = (AttrNumber *)
-			palloc0(tupleDesc->natts * sizeof(AttrNumber));
+		newattmap = make_attrmap(tupleDesc->natts);
 
 		for (parent_attno = 1; parent_attno <= tupleDesc->natts;
 			 parent_attno++)
@@ -2259,7 +2257,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 			 * Ignore dropped columns in the parent.
 			 */
 			if (attribute->attisdropped)
-				continue;		/* leave newattno entry as zero */
+				continue;		/* leave newattmap->attnums entry as zero */
 
 			/*
 			 * Does it conflict with some previously inherited column?
@@ -2317,7 +2315,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				/* Merge of NOT NULL constraints = OR 'em together */
 				def->is_not_null |= attribute->attnotnull;
 				/* Default and other constraints are handled below */
-				newattno[parent_attno - 1] = exist_attno;
+				newattmap->attnums[parent_attno - 1] = exist_attno;
 
 				/* Check for GENERATED conflicts */
 				if (def->generated != attribute->attgenerated)
@@ -2348,7 +2346,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				def->constraints = NIL;
 				def->location = -1;
 				inhSchema = lappend(inhSchema, def);
-				newattno[parent_attno - 1] = ++child_attno;
+				newattmap->attnums[parent_attno - 1] = ++child_attno;
 			}
 
 			/*
@@ -2396,7 +2394,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 
 		/*
 		 * Now copy the CHECK constraints of this parent, adjusting attnos
-		 * using the completed newattno[] map.  Identically named constraints
+		 * using the completed newattmap map.  Identically named constraints
 		 * are merged if possible, else we throw error.
 		 */
 		if (constr && constr->num_check > 0)
@@ -2417,7 +2415,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				/* Adjust Vars to match new table's column numbering */
 				expr = map_variable_attnos(stringToNode(check[i].ccbin),
 										   1, 0,
-										   newattno, tupleDesc->natts,
+										   newattmap,
 										   InvalidOid, &found_whole_row);
 
 				/*
@@ -2454,7 +2452,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 			}
 		}
 
-		pfree(newattno);
+		free_attrmap(newattmap);
 
 		/*
 		 * Close the parent rel, but keep our lock on it until xact commit.
@@ -8197,7 +8195,7 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel,
 		for (int i = 0; i < pd->nparts; i++)
 		{
 			Relation	partRel;
-			AttrNumber *map;
+			AttrMap    *map;
 			AttrNumber *mapped_pkattnum;
 			Oid			partIndexId;
 
@@ -8207,13 +8205,13 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel,
 			 * Map the attribute numbers in the referenced side of the FK
 			 * definition to match the partition's column layout.
 			 */
-			map = convert_tuples_by_name_map_if_req(RelationGetDescr(partRel),
-													RelationGetDescr(pkrel));
+			map = build_attrmap_by_name_if_req(RelationGetDescr(partRel),
+											   RelationGetDescr(pkrel));
 			if (map)
 			{
 				mapped_pkattnum = palloc(sizeof(AttrNumber) * numfks);
 				for (int j = 0; j < numfks; j++)
-					mapped_pkattnum[j] = map[pkattnum[j] - 1];
+					mapped_pkattnum[j] = map->attnums[pkattnum[j] - 1];
 			}
 			else
 				mapped_pkattnum = pkattnum;
@@ -8234,7 +8232,7 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel,
 			if (map)
 			{
 				pfree(mapped_pkattnum);
-				pfree(map);
+				free_attrmap(map);
 			}
 		}
 	}
@@ -8338,7 +8336,7 @@ addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel,
 			Oid			partitionId = pd->oids[i];
 			Relation	partition = table_open(partitionId, lockmode);
 			List	   *partFKs;
-			AttrNumber *attmap;
+			AttrMap    *attmap;
 			AttrNumber	mapped_fkattnum[INDEX_MAX_KEYS];
 			bool		attached;
 			char	   *conname;
@@ -8349,10 +8347,10 @@ addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel,
 
 			CheckTableNotInUse(partition, "ALTER TABLE");
 
-			attmap = convert_tuples_by_name_map(RelationGetDescr(partition),
-												RelationGetDescr(rel));
+			attmap = build_attrmap_by_name(RelationGetDescr(partition),
+										   RelationGetDescr(rel));
 			for (int j = 0; j < numfks; j++)
-				mapped_fkattnum[j] = attmap[fkattnum[j] - 1];
+				mapped_fkattnum[j] = attmap->attnums[fkattnum[j] - 1];
 
 			/* Check whether an existing constraint can be repurposed */
 			partFKs = copyObject(RelationGetFKeyList(partition));
@@ -8500,7 +8498,7 @@ static void
 CloneFkReferenced(Relation parentRel, Relation partitionRel)
 {
 	Relation	pg_constraint;
-	AttrNumber *attmap;
+	AttrMap    *attmap;
 	ListCell   *cell;
 	SysScanDesc scan;
 	ScanKeyData key[2];
@@ -8539,8 +8537,8 @@ CloneFkReferenced(Relation parentRel, Relation partitionRel)
 	systable_endscan(scan);
 	table_close(pg_constraint, RowShareLock);
 
-	attmap = convert_tuples_by_name_map(RelationGetDescr(partitionRel),
-										RelationGetDescr(parentRel));
+	attmap = build_attrmap_by_name(RelationGetDescr(partitionRel),
+								   RelationGetDescr(parentRel));
 	foreach(cell, clone)
 	{
 		Oid			constrOid = lfirst_oid(cell);
@@ -8578,8 +8576,9 @@ CloneFkReferenced(Relation parentRel, Relation partitionRel)
 								   conpfeqop,
 								   conppeqop,
 								   conffeqop);
+		Assert(numfks == attmap->maplen);
 		for (int i = 0; i < numfks; i++)
-			mapped_confkey[i] = attmap[confkey[i] - 1];
+			mapped_confkey[i] = attmap->attnums[confkey[i] - 1];
 
 		fkconstraint = makeNode(Constraint);
 		/* for now this is all we need */
@@ -8646,7 +8645,7 @@ CloneFkReferenced(Relation parentRel, Relation partitionRel)
 static void
 CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel)
 {
-	AttrNumber *attmap;
+	AttrMap    *attmap;
 	List	   *partFKs;
 	List	   *clone = NIL;
 	ListCell   *cell;
@@ -8675,8 +8674,8 @@ CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel)
 	 * The constraint key may differ, if the columns in the partition are
 	 * different.  This map is used to convert them.
 	 */
-	attmap = convert_tuples_by_name_map(RelationGetDescr(partRel),
-										RelationGetDescr(parentRel));
+	attmap = build_attrmap_by_name(RelationGetDescr(partRel),
+								   RelationGetDescr(parentRel));
 
 	partFKs = copyObject(RelationGetFKeyList(partRel));
 
@@ -8725,8 +8724,8 @@ CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel)
 
 		DeconstructFkConstraintRow(tuple, &numfks, conkey, confkey,
 								   conpfeqop, conppeqop, conffeqop);
-		for (int i = 0; i < numfks; i++)
-			mapped_conkey[i] = attmap[conkey[i] - 1];
+		for (int i = 0; i < attmap->maplen; i++)
+			mapped_conkey[i] = attmap->attnums[conkey[i] - 1];
 
 		/*
 		 * Before creating a new constraint, see whether any existing FKs are
@@ -10499,18 +10498,18 @@ ATPrepAlterColumnType(List **wqueue,
 			 */
 			if (def->cooked_default)
 			{
-				AttrNumber *attmap;
+				AttrMap    *attmap;
 				bool		found_whole_row;
 
 				/* create a copy to scribble on */
 				cmd = copyObject(cmd);
 
-				attmap = convert_tuples_by_name_map(RelationGetDescr(childrel),
-													RelationGetDescr(rel));
+				attmap = build_attrmap_by_name(RelationGetDescr(childrel),
+											   RelationGetDescr(rel));
 				((ColumnDef *) cmd->def)->cooked_default =
 					map_variable_attnos(def->cooked_default,
 										1, 0,
-										attmap, RelationGetDescr(rel)->natts,
+										attmap,
 										InvalidOid, &found_whole_row);
 				if (found_whole_row)
 					ereport(ERROR,
@@ -15862,7 +15861,7 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 		Oid			idx = lfirst_oid(cell);
 		Relation	idxRel = index_open(idx, AccessShareLock);
 		IndexInfo  *info;
-		AttrNumber *attmap;
+		AttrMap    *attmap;
 		bool		found = false;
 		Oid			constraintOid;
 
@@ -15878,8 +15877,8 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 
 		/* construct an indexinfo to compare existing indexes against */
 		info = BuildIndexInfo(idxRel);
-		attmap = convert_tuples_by_name_map(RelationGetDescr(attachrel),
-											RelationGetDescr(rel));
+		attmap = build_attrmap_by_name(RelationGetDescr(attachrel),
+									   RelationGetDescr(rel));
 		constraintOid = get_relation_idx_constraint_oid(RelationGetRelid(rel), idx);
 
 		/*
@@ -15901,8 +15900,7 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 								 idxRel->rd_indcollation,
 								 attachrelIdxRels[i]->rd_opfamily,
 								 idxRel->rd_opfamily,
-								 attmap,
-								 RelationGetDescr(rel)->natts))
+								 attmap))
 			{
 				/*
 				 * If this index is being created in the parent because of a
@@ -15943,7 +15941,6 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 
 			stmt = generateClonedIndexStmt(NULL,
 										   idxRel, attmap,
-										   RelationGetDescr(rel)->natts,
 										   &constraintOid);
 			DefineIndex(RelationGetRelid(attachrel), stmt, InvalidOid,
 						RelationGetRelid(idxRel),
@@ -16409,7 +16406,7 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name)
 	{
 		IndexInfo  *childInfo;
 		IndexInfo  *parentInfo;
-		AttrNumber *attmap;
+		AttrMap    *attmap;
 		bool		found;
 		int			i;
 		PartitionDesc partDesc;
@@ -16454,15 +16451,14 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name)
 		/* Ensure the indexes are compatible */
 		childInfo = BuildIndexInfo(partIdx);
 		parentInfo = BuildIndexInfo(parentIdx);
-		attmap = convert_tuples_by_name_map(RelationGetDescr(partTbl),
-											RelationGetDescr(parentTbl));
+		attmap = build_attrmap_by_name(RelationGetDescr(partTbl),
+									   RelationGetDescr(parentTbl));
 		if (!CompareIndexInfo(childInfo, parentInfo,
 							  partIdx->rd_indcollation,
 							  parentIdx->rd_indcollation,
 							  partIdx->rd_opfamily,
 							  parentIdx->rd_opfamily,
-							  attmap,
-							  RelationGetDescr(parentTbl)->natts))
+							  attmap))
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
 					 errmsg("cannot attach index \"%s\" as a partition of index \"%s\"",
@@ -16499,7 +16495,7 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name)
 			ConstraintSetParentConstraint(cldConstrId, constraintOid,
 										  RelationGetRelid(partTbl));
 
-		pfree(attmap);
+		free_attrmap(attmap);
 
 		validatePartitionedIndex(parentIdx, parentTbl);
 	}
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index c46eb8d646..d60ea9c5af 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1843,14 +1843,14 @@ ExecPartitionCheckEmitError(ResultRelInfo *resultRelInfo,
 	if (resultRelInfo->ri_PartitionRoot)
 	{
 		TupleDesc	old_tupdesc;
-		AttrNumber *map;
+		AttrMap    *map;
 
 		root_relid = RelationGetRelid(resultRelInfo->ri_PartitionRoot);
 		tupdesc = RelationGetDescr(resultRelInfo->ri_PartitionRoot);
 
 		old_tupdesc = RelationGetDescr(resultRelInfo->ri_RelationDesc);
 		/* a reverse map */
-		map = convert_tuples_by_name_map_if_req(old_tupdesc, tupdesc);
+		map = build_attrmap_by_name_if_req(old_tupdesc, tupdesc);
 
 		/*
 		 * Partition-specific slot's tupdesc can't be changed, so allocate a
@@ -1929,13 +1929,13 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 				 */
 				if (resultRelInfo->ri_PartitionRoot)
 				{
-					AttrNumber *map;
+					AttrMap    *map;
 
 					rel = resultRelInfo->ri_PartitionRoot;
 					tupdesc = RelationGetDescr(rel);
 					/* a reverse map */
-					map = convert_tuples_by_name_map_if_req(orig_tupdesc,
-															tupdesc);
+					map = build_attrmap_by_name_if_req(orig_tupdesc,
+													   tupdesc);
 
 					/*
 					 * Partition-specific slot's tupdesc can't be changed, so
@@ -1978,13 +1978,13 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 			if (resultRelInfo->ri_PartitionRoot)
 			{
 				TupleDesc	old_tupdesc = RelationGetDescr(rel);
-				AttrNumber *map;
+				AttrMap    *map;
 
 				rel = resultRelInfo->ri_PartitionRoot;
 				tupdesc = RelationGetDescr(rel);
 				/* a reverse map */
-				map = convert_tuples_by_name_map_if_req(old_tupdesc,
-														tupdesc);
+				map = build_attrmap_by_name_if_req(old_tupdesc,
+												   tupdesc);
 
 				/*
 				 * Partition-specific slot's tupdesc can't be changed, so
@@ -2085,13 +2085,13 @@ ExecWithCheckOptions(WCOKind kind, ResultRelInfo *resultRelInfo,
 					if (resultRelInfo->ri_PartitionRoot)
 					{
 						TupleDesc	old_tupdesc = RelationGetDescr(rel);
-						AttrNumber *map;
+						AttrMap    *map;
 
 						rel = resultRelInfo->ri_PartitionRoot;
 						tupdesc = RelationGetDescr(rel);
 						/* a reverse map */
-						map = convert_tuples_by_name_map_if_req(old_tupdesc,
-																tupdesc);
+						map = build_attrmap_by_name_if_req(old_tupdesc,
+														   tupdesc);
 
 						/*
 						 * Partition-specific slot's tupdesc can't be changed,
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index d23f292cb0..d8cbd6a2f8 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -143,7 +143,7 @@ typedef struct PartitionDispatchData
 	List	   *keystate;		/* list of ExprState */
 	PartitionDesc partdesc;
 	TupleTableSlot *tupslot;
-	AttrNumber *tupmap;
+	AttrMap    *tupmap;
 	int			indexes[FLEXIBLE_ARRAY_MEMBER];
 }			PartitionDispatchData;
 
@@ -298,7 +298,7 @@ ExecFindPartition(ModifyTableState *mtstate,
 	dispatch = pd[0];
 	while (true)
 	{
-		AttrNumber *map = dispatch->tupmap;
+		AttrMap    *map = dispatch->tupmap;
 		int			partidx = -1;
 
 		CHECK_FOR_INTERRUPTS();
@@ -511,7 +511,7 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 	Relation	firstResultRel = mtstate->resultRelInfo[0].ri_RelationDesc;
 	ResultRelInfo *leaf_part_rri;
 	MemoryContext oldcxt;
-	AttrNumber *part_attnos = NULL;
+	AttrMap    *part_attmap = NULL;
 	bool		found_whole_row;
 
 	oldcxt = MemoryContextSwitchTo(proute->memcxt);
@@ -584,14 +584,13 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 		/*
 		 * Convert Vars in it to contain this partition's attribute numbers.
 		 */
-		part_attnos =
-			convert_tuples_by_name_map(RelationGetDescr(partrel),
-									   RelationGetDescr(firstResultRel));
+		part_attmap =
+			build_attrmap_by_name(RelationGetDescr(partrel),
+								  RelationGetDescr(firstResultRel));
 		wcoList = (List *)
 			map_variable_attnos((Node *) wcoList,
 								firstVarno, 0,
-								part_attnos,
-								RelationGetDescr(firstResultRel)->natts,
+								part_attmap,
 								RelationGetForm(partrel)->reltype,
 								&found_whole_row);
 		/* We ignore the value of found_whole_row. */
@@ -642,15 +641,14 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 		/*
 		 * Convert Vars in it to contain this partition's attribute numbers.
 		 */
-		if (part_attnos == NULL)
-			part_attnos =
-				convert_tuples_by_name_map(RelationGetDescr(partrel),
-										   RelationGetDescr(firstResultRel));
+		if (part_attmap == NULL)
+			part_attmap =
+				build_attrmap_by_name(RelationGetDescr(partrel),
+									  RelationGetDescr(firstResultRel));
 		returningList = (List *)
 			map_variable_attnos((Node *) returningList,
 								firstVarno, 0,
-								part_attnos,
-								RelationGetDescr(firstResultRel)->natts,
+								part_attmap,
 								RelationGetForm(partrel)->reltype,
 								&found_whole_row);
 		/* We ignore the value of found_whole_row. */
@@ -785,23 +783,21 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 				 * target relation (firstVarno).
 				 */
 				onconflset = (List *) copyObject((Node *) node->onConflictSet);
-				if (part_attnos == NULL)
-					part_attnos =
-						convert_tuples_by_name_map(RelationGetDescr(partrel),
-												   RelationGetDescr(firstResultRel));
+				if (part_attmap == NULL)
+					part_attmap =
+						build_attrmap_by_name(RelationGetDescr(partrel),
+											  RelationGetDescr(firstResultRel));
 				onconflset = (List *)
 					map_variable_attnos((Node *) onconflset,
 										INNER_VAR, 0,
-										part_attnos,
-										RelationGetDescr(firstResultRel)->natts,
+										part_attmap,
 										RelationGetForm(partrel)->reltype,
 										&found_whole_row);
 				/* We ignore the value of found_whole_row. */
 				onconflset = (List *)
 					map_variable_attnos((Node *) onconflset,
 										firstVarno, 0,
-										part_attnos,
-										RelationGetDescr(firstResultRel)->natts,
+										part_attmap,
 										RelationGetForm(partrel)->reltype,
 										&found_whole_row);
 				/* We ignore the value of found_whole_row. */
@@ -835,16 +831,14 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 					clause = (List *)
 						map_variable_attnos((Node *) clause,
 											INNER_VAR, 0,
-											part_attnos,
-											RelationGetDescr(firstResultRel)->natts,
+											part_attmap,
 											RelationGetForm(partrel)->reltype,
 											&found_whole_row);
 					/* We ignore the value of found_whole_row. */
 					clause = (List *)
 						map_variable_attnos((Node *) clause,
 											firstVarno, 0,
-											part_attnos,
-											RelationGetDescr(firstResultRel)->natts,
+											part_attmap,
 											RelationGetForm(partrel)->reltype,
 											&found_whole_row);
 					/* We ignore the value of found_whole_row. */
@@ -1036,8 +1030,8 @@ ExecInitPartitionDispatchInfo(EState *estate,
 		 * tuple descriptor when computing its partition key for tuple
 		 * routing.
 		 */
-		pd->tupmap = convert_tuples_by_name_map_if_req(RelationGetDescr(parent_pd->reldesc),
-													   tupdesc);
+		pd->tupmap = build_attrmap_by_name_if_req(RelationGetDescr(parent_pd->reldesc),
+												  tupdesc);
 		pd->tupslot = pd->tupmap ?
 			MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual) : NULL;
 	}
@@ -1434,15 +1428,16 @@ adjust_partition_tlist(List *tlist, TupleConversionMap *map)
 {
 	List	   *new_tlist = NIL;
 	TupleDesc	tupdesc = map->outdesc;
-	AttrNumber *attrMap = map->attrMap;
+	AttrMap    *attrMap = map->attrMap;
 	AttrNumber	attrno;
 
+	Assert(tupdesc->natts == attrMap->maplen);
 	for (attrno = 1; attrno <= tupdesc->natts; attrno++)
 	{
 		Form_pg_attribute att_tup = TupleDescAttr(tupdesc, attrno - 1);
 		TargetEntry *tle;
 
-		if (attrMap[attrno - 1] != InvalidAttrNumber)
+		if (attrMap->attnums[attrno - 1] != InvalidAttrNumber)
 		{
 			Assert(!att_tup->attisdropped);
 
@@ -1450,7 +1445,7 @@ adjust_partition_tlist(List *tlist, TupleConversionMap *map)
 			 * Use the corresponding entry from the parent's tlist, adjusting
 			 * the resno the match the partition's attno.
 			 */
-			tle = (TargetEntry *) list_nth(tlist, attrMap[attrno - 1] - 1);
+			tle = (TargetEntry *) list_nth(tlist, attrMap->attnums[attrno - 1] - 1);
 			tle->resno = attrno;
 		}
 		else
diff --git a/src/backend/jit/llvm/llvmjit_expr.c b/src/backend/jit/llvm/llvmjit_expr.c
index a9d362100a..ffd887c71a 100644
--- a/src/backend/jit/llvm/llvmjit_expr.c
+++ b/src/backend/jit/llvm/llvmjit_expr.c
@@ -20,7 +20,6 @@
 
 #include "access/htup_details.h"
 #include "access/nbtree.h"
-#include "access/tupconvert.h"
 #include "catalog/objectaccess.h"
 #include "catalog/pg_type.h"
 #include "executor/execExpr.h"
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index ee47547624..d1b9a8accc 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -917,7 +917,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 	Relation	relation;
 	TupleDesc	tupleDesc;
 	TupleConstr *constr;
-	AttrNumber *attmap;
+	AttrMap    *attmap;
 	AclResult	aclresult;
 	char	   *comment;
 	ParseCallbackState pcbstate;
@@ -974,7 +974,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 	 * since dropped columns in the source table aren't copied, so the new
 	 * table can have different column numbers.
 	 */
-	attmap = (AttrNumber *) palloc0(sizeof(AttrNumber) * tupleDesc->natts);
+	attmap = make_attrmap(tupleDesc->natts);
 
 	/*
 	 * Insert the copied attributes into the cxt for the new table definition.
@@ -1020,7 +1020,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 		 */
 		cxt->columns = lappend(cxt->columns, def);
 
-		attmap[parent_attno - 1] = list_length(cxt->columns);
+		attmap->attnums[parent_attno - 1] = list_length(cxt->columns);
 
 		/*
 		 * Copy default, if present and it should be copied.  We have separate
@@ -1051,7 +1051,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 
 			def->cooked_default = map_variable_attnos(this_default,
 													  1, 0,
-													  attmap, tupleDesc->natts,
+													  attmap,
 													  InvalidOid, &found_whole_row);
 
 			/*
@@ -1134,7 +1134,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 
 			ccbin_node = map_variable_attnos(stringToNode(ccbin),
 											 1, 0,
-											 attmap, tupleDesc->natts,
+											 attmap,
 											 InvalidOid, &found_whole_row);
 
 			/*
@@ -1200,7 +1200,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 			/* Build CREATE INDEX statement to recreate the parent_index */
 			index_stmt = generateClonedIndexStmt(cxt->relation,
 												 parent_index,
-												 attmap, tupleDesc->natts,
+												 attmap,
 												 NULL);
 
 			/* Copy comment on index, if requested */
@@ -1332,7 +1332,7 @@ transformOfType(CreateStmtContext *cxt, TypeName *ofTypename)
  */
 IndexStmt *
 generateClonedIndexStmt(RangeVar *heapRel, Relation source_idx,
-						const AttrNumber *attmap, int attmap_length,
+						const AttrMap *attmap,
 						Oid *constraintOid)
 {
 	Oid			source_relid = RelationGetRelid(source_idx);
@@ -1552,7 +1552,7 @@ generateClonedIndexStmt(RangeVar *heapRel, Relation source_idx,
 			/* Adjust Vars to match new table's column numbering */
 			indexkey = map_variable_attnos(indexkey,
 										   1, 0,
-										   attmap, attmap_length,
+										   attmap,
 										   InvalidOid, &found_whole_row);
 
 			/* As in transformTableLikeClause, reject whole-row variables */
@@ -1659,7 +1659,7 @@ generateClonedIndexStmt(RangeVar *heapRel, Relation source_idx,
 		/* Adjust Vars to match new table's column numbering */
 		pred_tree = map_variable_attnos(pred_tree,
 										1, 0,
-										attmap, attmap_length,
+										attmap,
 										InvalidOid, &found_whole_row);
 
 		/* As in transformTableLikeClause, reject whole-row variables */
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index b386f8460d..f38c5b3ea4 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -267,7 +267,7 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
 		 */
 		desc = RelationGetDescr(entry->localrel);
 		oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
-		entry->attrmap = palloc(desc->natts * sizeof(AttrNumber));
+		entry->attrmap = make_attrmap(desc->natts);
 		MemoryContextSwitchTo(oldctx);
 
 		found = 0;
@@ -278,14 +278,14 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
 
 			if (attr->attisdropped || attr->attgenerated)
 			{
-				entry->attrmap[i] = -1;
+				entry->attrmap->attnums[i] = -1;
 				continue;
 			}
 
 			attnum = logicalrep_rel_att_by_name(remoterel,
 												NameStr(attr->attname));
 
-			entry->attrmap[i] = attnum;
+			entry->attrmap->attnums[i] = attnum;
 			if (attnum >= 0)
 				found++;
 		}
@@ -340,8 +340,8 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
 
 			attnum = AttrNumberGetAttrOffset(attnum);
 
-			if (entry->attrmap[attnum] < 0 ||
-				!bms_is_member(entry->attrmap[attnum], remoterel->attkeys))
+			if (entry->attrmap->attnums[attnum] < 0 ||
+				!bms_is_member(entry->attrmap->attnums[attnum], remoterel->attkeys))
 			{
 				entry->updatable = false;
 				break;
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index ced0d191c2..b1d783383f 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -232,6 +232,7 @@ slot_fill_defaults(LogicalRepRelMapEntry *rel, EState *estate,
 	defmap = (int *) palloc(num_phys_attrs * sizeof(int));
 	defexprs = (ExprState **) palloc(num_phys_attrs * sizeof(ExprState *));
 
+	Assert(rel->attrmap->maplen == num_phys_attrs);
 	for (attnum = 0; attnum < num_phys_attrs; attnum++)
 	{
 		Expr	   *defexpr;
@@ -239,7 +240,7 @@ slot_fill_defaults(LogicalRepRelMapEntry *rel, EState *estate,
 		if (TupleDescAttr(desc, attnum)->attisdropped || TupleDescAttr(desc, attnum)->attgenerated)
 			continue;
 
-		if (rel->attrmap[attnum] >= 0)
+		if (rel->attrmap->attnums[attnum] >= 0)
 			continue;
 
 		defexpr = (Expr *) build_column_default(rel->localrel, attnum + 1);
@@ -321,10 +322,11 @@ slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel,
 	error_context_stack = &errcallback;
 
 	/* Call the "in" function for each non-dropped attribute */
+	Assert(natts == rel->attrmap->maplen);
 	for (i = 0; i < natts; i++)
 	{
 		Form_pg_attribute att = TupleDescAttr(slot->tts_tupleDescriptor, i);
-		int			remoteattnum = rel->attrmap[i];
+		int			remoteattnum = rel->attrmap->attnums[i];
 
 		if (!att->attisdropped && remoteattnum >= 0 &&
 			values[remoteattnum] != NULL)
@@ -405,10 +407,11 @@ slot_modify_cstrings(TupleTableSlot *slot, TupleTableSlot *srcslot,
 	error_context_stack = &errcallback;
 
 	/* Call the "in" function for each replaced attribute */
+	Assert(natts == rel->attrmap->maplen);
 	for (i = 0; i < natts; i++)
 	{
 		Form_pg_attribute att = TupleDescAttr(slot->tts_tupleDescriptor, i);
-		int			remoteattnum = rel->attrmap[i];
+		int			remoteattnum = rel->attrmap->attnums[i];
 
 		if (remoteattnum < 0)
 			continue;
diff --git a/src/backend/rewrite/rewriteManip.c b/src/backend/rewrite/rewriteManip.c
index 93508c2a87..12f7cadf3b 100644
--- a/src/backend/rewrite/rewriteManip.c
+++ b/src/backend/rewrite/rewriteManip.c
@@ -1221,8 +1221,7 @@ typedef struct
 {
 	int			target_varno;	/* RTE index to search for */
 	int			sublevels_up;	/* (current) nesting depth */
-	const AttrNumber *attno_map;	/* map array for user attnos */
-	int			map_length;		/* number of entries in attno_map[] */
+	const AttrMap *attno_map;	/* map array for user attnos */
 	Oid			to_rowtype;		/* change whole-row Vars to this type */
 	bool	   *found_whole_row;	/* output flag */
 } map_variable_attnos_context;
@@ -1249,11 +1248,11 @@ map_variable_attnos_mutator(Node *node,
 			if (attno > 0)
 			{
 				/* user-defined column, replace attno */
-				if (attno > context->map_length ||
-					context->attno_map[attno - 1] == 0)
+				if (attno > context->attno_map->maplen ||
+					context->attno_map->attnums[attno - 1] == 0)
 					elog(ERROR, "unexpected varattno %d in expression to be mapped",
 						 attno);
-				newvar->varattno = newvar->varoattno = context->attno_map[attno - 1];
+				newvar->varattno = newvar->varoattno = context->attno_map->attnums[attno - 1];
 			}
 			else if (attno == 0)
 			{
@@ -1350,7 +1349,7 @@ map_variable_attnos_mutator(Node *node,
 Node *
 map_variable_attnos(Node *node,
 					int target_varno, int sublevels_up,
-					const AttrNumber *attno_map, int map_length,
+					const AttrMap *attno_map,
 					Oid to_rowtype, bool *found_whole_row)
 {
 	map_variable_attnos_context context;
@@ -1358,7 +1357,6 @@ map_variable_attnos(Node *node,
 	context.target_varno = target_varno;
 	context.sublevels_up = sublevels_up;
 	context.attno_map = attno_map;
-	context.map_length = map_length;
 	context.to_rowtype = to_rowtype;
 	context.found_whole_row = found_whole_row;
 
diff --git a/src/include/access/attmap.h b/src/include/access/attmap.h
new file mode 100644
index 0000000000..f71afbe72b
--- /dev/null
+++ b/src/include/access/attmap.h
@@ -0,0 +1,51 @@
+/*-------------------------------------------------------------------------
+ *
+ * attmap.h
+ *	  Definitions for PostgreSQL attribute mappings
+ *
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/attmap.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef ATTMAP_H
+#define ATTMAP_H
+
+#include "access/attnum.h"
+#include "access/tupdesc.h"
+
+/*
+ * Attribute mapping structure
+ *
+ * An attribute mapping tracks the relationship of a child relation and
+ * its parent for inheritance and partitions.  This is used mainly for
+ * cloned object creations (indexes, foreign keys, etc.) when creating
+ * an inherited child relation, and for runtime-execution attribute
+ * mapping.
+ *
+ * Dropped attributes are marked with 0 and the length of the map is set
+ * to be the number of attributes of the parent, which takes into account
+ * its dropped attributes.
+ */
+typedef struct AttrMap
+{
+	AttrNumber *attnums;
+	int			maplen;
+} AttrMap;
+
+extern AttrMap *make_attrmap(int maplen);
+extern void free_attrmap(AttrMap *map);
+
+/* Convertion routines to build mappings */
+extern AttrMap *build_attrmap_by_name(TupleDesc indesc,
+									  TupleDesc outdesc);
+extern AttrMap *build_attrmap_by_name_if_req(TupleDesc indesc,
+											 TupleDesc outdesc);
+extern AttrMap *build_attrmap_by_position(TupleDesc indesc,
+										  TupleDesc outdesc,
+										  const char *msg);
+
+#endif							/* ATTMAP_H */
diff --git a/src/include/access/tupconvert.h b/src/include/access/tupconvert.h
index 6d095f8e0d..5ed7fe6d39 100644
--- a/src/include/access/tupconvert.h
+++ b/src/include/access/tupconvert.h
@@ -14,6 +14,7 @@
 #ifndef TUPCONVERT_H
 #define TUPCONVERT_H
 
+#include "access/attmap.h"
 #include "access/htup.h"
 #include "access/tupdesc.h"
 #include "executor/tuptable.h"
@@ -23,7 +24,7 @@ typedef struct TupleConversionMap
 {
 	TupleDesc	indesc;			/* tupdesc for source rowtype */
 	TupleDesc	outdesc;		/* tupdesc for result rowtype */
-	AttrNumber *attrMap;		/* indexes of input fields, or 0 for null */
+	AttrMap    *attrMap;		/* indexes of input fields, or 0 for null */
 	Datum	   *invalues;		/* workspace for deconstructing source */
 	bool	   *inisnull;
 	Datum	   *outvalues;		/* workspace for constructing result */
@@ -38,14 +39,10 @@ extern TupleConversionMap *convert_tuples_by_position(TupleDesc indesc,
 extern TupleConversionMap *convert_tuples_by_name(TupleDesc indesc,
 												  TupleDesc outdesc);
 
-extern AttrNumber *convert_tuples_by_name_map(TupleDesc indesc,
-											  TupleDesc outdesc);
-extern AttrNumber *convert_tuples_by_name_map_if_req(TupleDesc indesc,
-													 TupleDesc outdesc);
-
 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 TupleTableSlot *execute_attr_map_slot(AttrMap *attrMap,
+											 TupleTableSlot *in_slot,
+											 TupleTableSlot *out_slot);
 
 extern void free_conversion_map(TupleConversionMap *map);
 
diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h
index 27d9e537d3..ea4ad1a2d1 100644
--- a/src/include/catalog/index.h
+++ b/src/include/catalog/index.h
@@ -111,7 +111,7 @@ extern IndexInfo *BuildDummyIndexInfo(Relation index);
 extern bool CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
 							 Oid *collations1, Oid *collations2,
 							 Oid *opfamilies1, Oid *opfamilies2,
-							 AttrNumber *attmap, int maplen);
+							 AttrMap *attmap);
 
 extern void BuildSpeculativeIndexInfo(Relation index, IndexInfo *ii);
 
diff --git a/src/include/parser/parse_utilcmd.h b/src/include/parser/parse_utilcmd.h
index 1348064ad0..08dd0ce4ca 100644
--- a/src/include/parser/parse_utilcmd.h
+++ b/src/include/parser/parse_utilcmd.h
@@ -16,6 +16,7 @@
 
 #include "parser/parse_node.h"
 
+typedef struct AttrMap AttrMap;
 
 extern List *transformCreateStmt(CreateStmt *stmt, const char *queryString);
 extern List *transformAlterTableStmt(Oid relid, AlterTableStmt *stmt,
@@ -29,7 +30,7 @@ extern PartitionBoundSpec *transformPartitionBound(ParseState *pstate, Relation
 												   PartitionBoundSpec *spec);
 extern IndexStmt *generateClonedIndexStmt(RangeVar *heapRel,
 										  Relation source_idx,
-										  const AttrNumber *attmap, int attmap_length,
+										  const AttrMap *attmap,
 										  Oid *constraintOid);
 
 #endif							/* PARSE_UTILCMD_H */
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index 2642a3f94e..950d99bfad 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -21,7 +21,7 @@ typedef struct LogicalRepRelMapEntry
 	/* Mapping to local relation, filled as needed. */
 	Oid			localreloid;	/* local relation id */
 	Relation	localrel;		/* relcache entry */
-	AttrNumber *attrmap;		/* map of local attributes to remote ones */
+	AttrMap    *attrmap;		/* map of local attributes to remote ones */
 	bool		updatable;		/* Can apply updates/deletes? */
 
 	/* Sync state. */
diff --git a/src/include/rewrite/rewriteManip.h b/src/include/rewrite/rewriteManip.h
index 8d8fd17e41..634cdc235d 100644
--- a/src/include/rewrite/rewriteManip.h
+++ b/src/include/rewrite/rewriteManip.h
@@ -17,6 +17,7 @@
 #include "nodes/parsenodes.h"
 
 
+typedef struct AttrMap AttrMap;
 typedef struct replace_rte_variables_context replace_rte_variables_context;
 
 typedef Node *(*replace_rte_variables_callback) (Var *var,
@@ -71,7 +72,7 @@ extern Node *replace_rte_variables_mutator(Node *node,
 
 extern Node *map_variable_attnos(Node *node,
 								 int target_varno, int sublevels_up,
-								 const AttrNumber *attno_map, int map_length,
+								 const AttrMap *attno_map,
 								 Oid to_rowtype, bool *found_whole_row);
 
 extern Node *ReplaceVarsFromTargetList(Node *node,
-- 
2.24.0

From 73bc37de86640b5e5df682d142d404ebba056b84 Mon Sep 17 00:00:00 2001
From: Michael Paquier <michael@paquier.xyz>
Date: Mon, 9 Dec 2019 11:51:33 +0900
Subject: [PATCH v3] Rework attribute mappings

---
 src/backend/access/common/Makefile         |   1 +
 src/backend/access/common/attmap.c         | 320 +++++++++++++++++++++
 src/backend/access/common/tupconvert.c     | 287 ++----------------
 src/backend/catalog/index.c                |  12 +-
 src/backend/catalog/partition.c            |  11 +-
 src/backend/catalog/pg_constraint.c        |   1 -
 src/backend/commands/indexcmds.c           |  16 +-
 src/backend/commands/tablecmds.c           |  98 +++----
 src/backend/executor/execMain.c            |  22 +-
 src/backend/executor/execPartition.c       |  57 ++--
 src/backend/jit/llvm/llvmjit_expr.c        |   1 -
 src/backend/parser/parse_utilcmd.c         |  18 +-
 src/backend/replication/logical/relation.c |  10 +-
 src/backend/replication/logical/worker.c   |   9 +-
 src/backend/rewrite/rewriteManip.c         |  12 +-
 src/include/access/attmap.h                |  52 ++++
 src/include/access/tupconvert.h            |  13 +-
 src/include/catalog/index.h                |   2 +-
 src/include/parser/parse_utilcmd.h         |   3 +-
 src/include/replication/logicalrelation.h  |   2 +-
 src/include/rewrite/rewriteManip.h         |   3 +-
 21 files changed, 531 insertions(+), 419 deletions(-)
 create mode 100644 src/backend/access/common/attmap.c
 create mode 100644 src/include/access/attmap.h

diff --git a/src/backend/access/common/Makefile b/src/backend/access/common/Makefile
index 6c9c6f3256..fd74e14024 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 0000000000..eebb7e76d7
--- /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 convert fully
+ * transformed expression trees from parent rowtype to child rowtype or
+ * vice-versa.  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 0ec9cd5870..85b194f394 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)
 	{
@@ -235,158 +132,13 @@ convert_tuples_by_name(TupleDesc indesc,
 	return map;
 }
 
-/*
- * 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);
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index e9955707fa..78b45ae6e1 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -2382,13 +2382,13 @@ BuildDummyIndexInfo(Relation index)
  * Note: passing collations and opfamilies separately is a kludge.  Adding
  * them to IndexInfo may result in better coding here and elsewhere.
  *
- * Use convert_tuples_by_name_map(index2, index1) to build the attmap.
+ * Use build_attrmap_by_name(index2, index1) to build the attmap.
  */
 bool
 CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
 				 Oid *collations1, Oid *collations2,
 				 Oid *opfamilies1, Oid *opfamilies2,
-				 AttrNumber *attmap, int maplen)
+				 AttrMap *attmap)
 {
 	int			i;
 
@@ -2415,12 +2415,12 @@ CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
 	 */
 	for (i = 0; i < info1->ii_NumIndexAttrs; i++)
 	{
-		if (maplen < info2->ii_IndexAttrNumbers[i])
+		if (attmap->maplen < info2->ii_IndexAttrNumbers[i])
 			elog(ERROR, "incorrect attribute map");
 
 		/* ignore expressions at this stage */
 		if ((info1->ii_IndexAttrNumbers[i] != InvalidAttrNumber) &&
-			(attmap[info2->ii_IndexAttrNumbers[i] - 1] !=
+			(attmap->attnums[info2->ii_IndexAttrNumbers[i] - 1] !=
 			 info1->ii_IndexAttrNumbers[i]))
 			return false;
 
@@ -2446,7 +2446,7 @@ CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
 		Node	   *mapped;
 
 		mapped = map_variable_attnos((Node *) info2->ii_Expressions,
-									 1, 0, attmap, maplen,
+									 1, 0, attmap,
 									 InvalidOid, &found_whole_row);
 		if (found_whole_row)
 		{
@@ -2470,7 +2470,7 @@ CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
 		Node	   *mapped;
 
 		mapped = map_variable_attnos((Node *) info2->ii_Predicate,
-									 1, 0, attmap, maplen,
+									 1, 0, attmap,
 									 InvalidOid, &found_whole_row);
 		if (found_whole_row)
 		{
diff --git a/src/backend/catalog/partition.c b/src/backend/catalog/partition.c
index 5dfa4499fd..7657608dd7 100644
--- a/src/backend/catalog/partition.c
+++ b/src/backend/catalog/partition.c
@@ -14,11 +14,11 @@
 */
 #include "postgres.h"
 
+#include "access/attmap.h"
 #include "access/genam.h"
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
-#include "access/tupconvert.h"
 #include "catalog/indexing.h"
 #include "catalog/partition.h"
 #include "catalog/pg_inherits.h"
@@ -206,14 +206,13 @@ map_partition_varattnos(List *expr, int fromrel_varno,
 
 	if (expr != NIL)
 	{
-		AttrNumber *part_attnos;
+		AttrMap    *part_attmap;
 
-		part_attnos = convert_tuples_by_name_map(RelationGetDescr(to_rel),
-												 RelationGetDescr(from_rel));
+		part_attmap = build_attrmap_by_name(RelationGetDescr(to_rel),
+											RelationGetDescr(from_rel));
 		expr = (List *) map_variable_attnos((Node *) expr,
 											fromrel_varno, 0,
-											part_attnos,
-											RelationGetDescr(from_rel)->natts,
+											part_attmap,
 											RelationGetForm(to_rel)->reltype,
 											&my_found_whole_row);
 	}
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index 56568b0105..25c52134f0 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -18,7 +18,6 @@
 #include "access/htup_details.h"
 #include "access/sysattr.h"
 #include "access/table.h"
-#include "access/tupconvert.h"
 #include "access/xact.h"
 #include "catalog/catalog.h"
 #include "catalog/dependency.h"
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index 374e2d0efe..8f242aef1e 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -1060,9 +1060,8 @@ DefineIndex(Oid relationId,
 				Relation	childrel;
 				List	   *childidxs;
 				ListCell   *cell;
-				AttrNumber *attmap;
+				AttrMap    *attmap;
 				bool		found = false;
-				int			maplen;
 
 				childrel = table_open(childRelid, lockmode);
 
@@ -1087,9 +1086,8 @@ DefineIndex(Oid relationId,
 
 				childidxs = RelationGetIndexList(childrel);
 				attmap =
-					convert_tuples_by_name_map(RelationGetDescr(childrel),
-											   parentDesc);
-				maplen = parentDesc->natts;
+					build_attrmap_by_name(RelationGetDescr(childrel),
+										  parentDesc);
 
 				foreach(cell, childidxs)
 				{
@@ -1108,7 +1106,7 @@ DefineIndex(Oid relationId,
 										 collationObjectId,
 										 cldidx->rd_opfamily,
 										 opfamOids,
-										 attmap, maplen))
+										 attmap))
 					{
 						Oid			cldConstrOid = InvalidOid;
 
@@ -1193,7 +1191,7 @@ DefineIndex(Oid relationId,
 						{
 							ielem->expr =
 								map_variable_attnos((Node *) ielem->expr,
-													1, 0, attmap, maplen,
+													1, 0, attmap,
 													InvalidOid,
 													&found_whole_row);
 							if (found_whole_row)
@@ -1202,7 +1200,7 @@ DefineIndex(Oid relationId,
 					}
 					childStmt->whereClause =
 						map_variable_attnos(stmt->whereClause, 1, 0,
-											attmap, maplen,
+											attmap,
 											InvalidOid, &found_whole_row);
 					if (found_whole_row)
 						elog(ERROR, "cannot convert whole-row table reference");
@@ -1217,7 +1215,7 @@ DefineIndex(Oid relationId,
 
 				pgstat_progress_update_param(PROGRESS_CREATEIDX_PARTITIONS_DONE,
 											 i + 1);
-				pfree(attmap);
+				free_attrmap(attmap);
 			}
 
 			/*
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 5440eb9015..2fbbcb9014 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -14,6 +14,7 @@
  */
 #include "postgres.h"
 
+#include "access/attmap.h"
 #include "access/genam.h"
 #include "access/heapam.h"
 #include "access/heapam_xlog.h"
@@ -22,7 +23,6 @@
 #include "access/relscan.h"
 #include "access/sysattr.h"
 #include "access/tableam.h"
-#include "access/tupconvert.h"
 #include "access/xact.h"
 #include "access/xlog.h"
 #include "catalog/catalog.h"
@@ -1072,7 +1072,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		foreach(cell, idxlist)
 		{
 			Relation	idxRel = index_open(lfirst_oid(cell), AccessShareLock);
-			AttrNumber *attmap;
+			AttrMap    *attmap;
 			IndexStmt  *idxstmt;
 			Oid			constraintOid;
 
@@ -1092,12 +1092,11 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 				}
 			}
 
-			attmap = convert_tuples_by_name_map(RelationGetDescr(rel),
-												RelationGetDescr(parent));
+			attmap = build_attrmap_by_name(RelationGetDescr(rel),
+										   RelationGetDescr(parent));
 			idxstmt =
 				generateClonedIndexStmt(NULL, idxRel,
-										attmap, RelationGetDescr(parent)->natts,
-										&constraintOid);
+										attmap, &constraintOid);
 			DefineIndex(RelationGetRelid(rel),
 						idxstmt,
 						InvalidOid,
@@ -2158,7 +2157,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 		Relation	relation;
 		TupleDesc	tupleDesc;
 		TupleConstr *constr;
-		AttrNumber *newattno;
+		AttrMap    *newattmap;
 		AttrNumber	parent_attno;
 
 		/* caller already got lock */
@@ -2239,12 +2238,11 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 		constr = tupleDesc->constr;
 
 		/*
-		 * newattno[] will contain the child-table attribute numbers for the
-		 * attributes of this parent table.  (They are not the same for
-		 * parents after the first one, nor if we have dropped columns.)
+		 * newattmap->attnums[] will contain the child-table attribute numbers
+		 * for the attributes of this parent table.  (They are not the same
+		 * for parents after the first one, nor if we have dropped columns.)
 		 */
-		newattno = (AttrNumber *)
-			palloc0(tupleDesc->natts * sizeof(AttrNumber));
+		newattmap = make_attrmap(tupleDesc->natts);
 
 		for (parent_attno = 1; parent_attno <= tupleDesc->natts;
 			 parent_attno++)
@@ -2259,7 +2257,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 			 * Ignore dropped columns in the parent.
 			 */
 			if (attribute->attisdropped)
-				continue;		/* leave newattno entry as zero */
+				continue;		/* leave newattmap->attnums entry as zero */
 
 			/*
 			 * Does it conflict with some previously inherited column?
@@ -2317,7 +2315,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				/* Merge of NOT NULL constraints = OR 'em together */
 				def->is_not_null |= attribute->attnotnull;
 				/* Default and other constraints are handled below */
-				newattno[parent_attno - 1] = exist_attno;
+				newattmap->attnums[parent_attno - 1] = exist_attno;
 
 				/* Check for GENERATED conflicts */
 				if (def->generated != attribute->attgenerated)
@@ -2348,7 +2346,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				def->constraints = NIL;
 				def->location = -1;
 				inhSchema = lappend(inhSchema, def);
-				newattno[parent_attno - 1] = ++child_attno;
+				newattmap->attnums[parent_attno - 1] = ++child_attno;
 			}
 
 			/*
@@ -2396,7 +2394,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 
 		/*
 		 * Now copy the CHECK constraints of this parent, adjusting attnos
-		 * using the completed newattno[] map.  Identically named constraints
+		 * using the completed newattmap map.  Identically named constraints
 		 * are merged if possible, else we throw error.
 		 */
 		if (constr && constr->num_check > 0)
@@ -2417,7 +2415,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 				/* Adjust Vars to match new table's column numbering */
 				expr = map_variable_attnos(stringToNode(check[i].ccbin),
 										   1, 0,
-										   newattno, tupleDesc->natts,
+										   newattmap,
 										   InvalidOid, &found_whole_row);
 
 				/*
@@ -2454,7 +2452,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
 			}
 		}
 
-		pfree(newattno);
+		free_attrmap(newattmap);
 
 		/*
 		 * Close the parent rel, but keep our lock on it until xact commit.
@@ -8197,7 +8195,7 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel,
 		for (int i = 0; i < pd->nparts; i++)
 		{
 			Relation	partRel;
-			AttrNumber *map;
+			AttrMap    *map;
 			AttrNumber *mapped_pkattnum;
 			Oid			partIndexId;
 
@@ -8207,13 +8205,13 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel,
 			 * Map the attribute numbers in the referenced side of the FK
 			 * definition to match the partition's column layout.
 			 */
-			map = convert_tuples_by_name_map_if_req(RelationGetDescr(partRel),
-													RelationGetDescr(pkrel));
+			map = build_attrmap_by_name_if_req(RelationGetDescr(partRel),
+											   RelationGetDescr(pkrel));
 			if (map)
 			{
 				mapped_pkattnum = palloc(sizeof(AttrNumber) * numfks);
 				for (int j = 0; j < numfks; j++)
-					mapped_pkattnum[j] = map[pkattnum[j] - 1];
+					mapped_pkattnum[j] = map->attnums[pkattnum[j] - 1];
 			}
 			else
 				mapped_pkattnum = pkattnum;
@@ -8234,7 +8232,7 @@ addFkRecurseReferenced(List **wqueue, Constraint *fkconstraint, Relation rel,
 			if (map)
 			{
 				pfree(mapped_pkattnum);
-				pfree(map);
+				free_attrmap(map);
 			}
 		}
 	}
@@ -8338,7 +8336,7 @@ addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel,
 			Oid			partitionId = pd->oids[i];
 			Relation	partition = table_open(partitionId, lockmode);
 			List	   *partFKs;
-			AttrNumber *attmap;
+			AttrMap    *attmap;
 			AttrNumber	mapped_fkattnum[INDEX_MAX_KEYS];
 			bool		attached;
 			char	   *conname;
@@ -8349,10 +8347,10 @@ addFkRecurseReferencing(List **wqueue, Constraint *fkconstraint, Relation rel,
 
 			CheckTableNotInUse(partition, "ALTER TABLE");
 
-			attmap = convert_tuples_by_name_map(RelationGetDescr(partition),
-												RelationGetDescr(rel));
+			attmap = build_attrmap_by_name(RelationGetDescr(partition),
+										   RelationGetDescr(rel));
 			for (int j = 0; j < numfks; j++)
-				mapped_fkattnum[j] = attmap[fkattnum[j] - 1];
+				mapped_fkattnum[j] = attmap->attnums[fkattnum[j] - 1];
 
 			/* Check whether an existing constraint can be repurposed */
 			partFKs = copyObject(RelationGetFKeyList(partition));
@@ -8500,7 +8498,7 @@ static void
 CloneFkReferenced(Relation parentRel, Relation partitionRel)
 {
 	Relation	pg_constraint;
-	AttrNumber *attmap;
+	AttrMap    *attmap;
 	ListCell   *cell;
 	SysScanDesc scan;
 	ScanKeyData key[2];
@@ -8539,8 +8537,8 @@ CloneFkReferenced(Relation parentRel, Relation partitionRel)
 	systable_endscan(scan);
 	table_close(pg_constraint, RowShareLock);
 
-	attmap = convert_tuples_by_name_map(RelationGetDescr(partitionRel),
-										RelationGetDescr(parentRel));
+	attmap = build_attrmap_by_name(RelationGetDescr(partitionRel),
+								   RelationGetDescr(parentRel));
 	foreach(cell, clone)
 	{
 		Oid			constrOid = lfirst_oid(cell);
@@ -8578,8 +8576,9 @@ CloneFkReferenced(Relation parentRel, Relation partitionRel)
 								   conpfeqop,
 								   conppeqop,
 								   conffeqop);
+		Assert(numfks == attmap->maplen);
 		for (int i = 0; i < numfks; i++)
-			mapped_confkey[i] = attmap[confkey[i] - 1];
+			mapped_confkey[i] = attmap->attnums[confkey[i] - 1];
 
 		fkconstraint = makeNode(Constraint);
 		/* for now this is all we need */
@@ -8646,7 +8645,7 @@ CloneFkReferenced(Relation parentRel, Relation partitionRel)
 static void
 CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel)
 {
-	AttrNumber *attmap;
+	AttrMap    *attmap;
 	List	   *partFKs;
 	List	   *clone = NIL;
 	ListCell   *cell;
@@ -8675,8 +8674,8 @@ CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel)
 	 * The constraint key may differ, if the columns in the partition are
 	 * different.  This map is used to convert them.
 	 */
-	attmap = convert_tuples_by_name_map(RelationGetDescr(partRel),
-										RelationGetDescr(parentRel));
+	attmap = build_attrmap_by_name(RelationGetDescr(partRel),
+								   RelationGetDescr(parentRel));
 
 	partFKs = copyObject(RelationGetFKeyList(partRel));
 
@@ -8726,7 +8725,7 @@ CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel)
 		DeconstructFkConstraintRow(tuple, &numfks, conkey, confkey,
 								   conpfeqop, conppeqop, conffeqop);
 		for (int i = 0; i < numfks; i++)
-			mapped_conkey[i] = attmap[conkey[i] - 1];
+			mapped_conkey[i] = attmap->attnums[conkey[i] - 1];
 
 		/*
 		 * Before creating a new constraint, see whether any existing FKs are
@@ -10499,18 +10498,18 @@ ATPrepAlterColumnType(List **wqueue,
 			 */
 			if (def->cooked_default)
 			{
-				AttrNumber *attmap;
+				AttrMap    *attmap;
 				bool		found_whole_row;
 
 				/* create a copy to scribble on */
 				cmd = copyObject(cmd);
 
-				attmap = convert_tuples_by_name_map(RelationGetDescr(childrel),
-													RelationGetDescr(rel));
+				attmap = build_attrmap_by_name(RelationGetDescr(childrel),
+											   RelationGetDescr(rel));
 				((ColumnDef *) cmd->def)->cooked_default =
 					map_variable_attnos(def->cooked_default,
 										1, 0,
-										attmap, RelationGetDescr(rel)->natts,
+										attmap,
 										InvalidOid, &found_whole_row);
 				if (found_whole_row)
 					ereport(ERROR,
@@ -15862,7 +15861,7 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 		Oid			idx = lfirst_oid(cell);
 		Relation	idxRel = index_open(idx, AccessShareLock);
 		IndexInfo  *info;
-		AttrNumber *attmap;
+		AttrMap    *attmap;
 		bool		found = false;
 		Oid			constraintOid;
 
@@ -15878,8 +15877,8 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 
 		/* construct an indexinfo to compare existing indexes against */
 		info = BuildIndexInfo(idxRel);
-		attmap = convert_tuples_by_name_map(RelationGetDescr(attachrel),
-											RelationGetDescr(rel));
+		attmap = build_attrmap_by_name(RelationGetDescr(attachrel),
+									   RelationGetDescr(rel));
 		constraintOid = get_relation_idx_constraint_oid(RelationGetRelid(rel), idx);
 
 		/*
@@ -15901,8 +15900,7 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 								 idxRel->rd_indcollation,
 								 attachrelIdxRels[i]->rd_opfamily,
 								 idxRel->rd_opfamily,
-								 attmap,
-								 RelationGetDescr(rel)->natts))
+								 attmap))
 			{
 				/*
 				 * If this index is being created in the parent because of a
@@ -15943,7 +15941,6 @@ AttachPartitionEnsureIndexes(Relation rel, Relation attachrel)
 
 			stmt = generateClonedIndexStmt(NULL,
 										   idxRel, attmap,
-										   RelationGetDescr(rel)->natts,
 										   &constraintOid);
 			DefineIndex(RelationGetRelid(attachrel), stmt, InvalidOid,
 						RelationGetRelid(idxRel),
@@ -16409,7 +16406,7 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name)
 	{
 		IndexInfo  *childInfo;
 		IndexInfo  *parentInfo;
-		AttrNumber *attmap;
+		AttrMap    *attmap;
 		bool		found;
 		int			i;
 		PartitionDesc partDesc;
@@ -16454,15 +16451,14 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name)
 		/* Ensure the indexes are compatible */
 		childInfo = BuildIndexInfo(partIdx);
 		parentInfo = BuildIndexInfo(parentIdx);
-		attmap = convert_tuples_by_name_map(RelationGetDescr(partTbl),
-											RelationGetDescr(parentTbl));
+		attmap = build_attrmap_by_name(RelationGetDescr(partTbl),
+									   RelationGetDescr(parentTbl));
 		if (!CompareIndexInfo(childInfo, parentInfo,
 							  partIdx->rd_indcollation,
 							  parentIdx->rd_indcollation,
 							  partIdx->rd_opfamily,
 							  parentIdx->rd_opfamily,
-							  attmap,
-							  RelationGetDescr(parentTbl)->natts))
+							  attmap))
 			ereport(ERROR,
 					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
 					 errmsg("cannot attach index \"%s\" as a partition of index \"%s\"",
@@ -16499,7 +16495,7 @@ ATExecAttachPartitionIdx(List **wqueue, Relation parentIdx, RangeVar *name)
 			ConstraintSetParentConstraint(cldConstrId, constraintOid,
 										  RelationGetRelid(partTbl));
 
-		pfree(attmap);
+		free_attrmap(attmap);
 
 		validatePartitionedIndex(parentIdx, parentTbl);
 	}
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index c46eb8d646..d60ea9c5af 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1843,14 +1843,14 @@ ExecPartitionCheckEmitError(ResultRelInfo *resultRelInfo,
 	if (resultRelInfo->ri_PartitionRoot)
 	{
 		TupleDesc	old_tupdesc;
-		AttrNumber *map;
+		AttrMap    *map;
 
 		root_relid = RelationGetRelid(resultRelInfo->ri_PartitionRoot);
 		tupdesc = RelationGetDescr(resultRelInfo->ri_PartitionRoot);
 
 		old_tupdesc = RelationGetDescr(resultRelInfo->ri_RelationDesc);
 		/* a reverse map */
-		map = convert_tuples_by_name_map_if_req(old_tupdesc, tupdesc);
+		map = build_attrmap_by_name_if_req(old_tupdesc, tupdesc);
 
 		/*
 		 * Partition-specific slot's tupdesc can't be changed, so allocate a
@@ -1929,13 +1929,13 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 				 */
 				if (resultRelInfo->ri_PartitionRoot)
 				{
-					AttrNumber *map;
+					AttrMap    *map;
 
 					rel = resultRelInfo->ri_PartitionRoot;
 					tupdesc = RelationGetDescr(rel);
 					/* a reverse map */
-					map = convert_tuples_by_name_map_if_req(orig_tupdesc,
-															tupdesc);
+					map = build_attrmap_by_name_if_req(orig_tupdesc,
+													   tupdesc);
 
 					/*
 					 * Partition-specific slot's tupdesc can't be changed, so
@@ -1978,13 +1978,13 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
 			if (resultRelInfo->ri_PartitionRoot)
 			{
 				TupleDesc	old_tupdesc = RelationGetDescr(rel);
-				AttrNumber *map;
+				AttrMap    *map;
 
 				rel = resultRelInfo->ri_PartitionRoot;
 				tupdesc = RelationGetDescr(rel);
 				/* a reverse map */
-				map = convert_tuples_by_name_map_if_req(old_tupdesc,
-														tupdesc);
+				map = build_attrmap_by_name_if_req(old_tupdesc,
+												   tupdesc);
 
 				/*
 				 * Partition-specific slot's tupdesc can't be changed, so
@@ -2085,13 +2085,13 @@ ExecWithCheckOptions(WCOKind kind, ResultRelInfo *resultRelInfo,
 					if (resultRelInfo->ri_PartitionRoot)
 					{
 						TupleDesc	old_tupdesc = RelationGetDescr(rel);
-						AttrNumber *map;
+						AttrMap    *map;
 
 						rel = resultRelInfo->ri_PartitionRoot;
 						tupdesc = RelationGetDescr(rel);
 						/* a reverse map */
-						map = convert_tuples_by_name_map_if_req(old_tupdesc,
-																tupdesc);
+						map = build_attrmap_by_name_if_req(old_tupdesc,
+														   tupdesc);
 
 						/*
 						 * Partition-specific slot's tupdesc can't be changed,
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index d23f292cb0..d8cbd6a2f8 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -143,7 +143,7 @@ typedef struct PartitionDispatchData
 	List	   *keystate;		/* list of ExprState */
 	PartitionDesc partdesc;
 	TupleTableSlot *tupslot;
-	AttrNumber *tupmap;
+	AttrMap    *tupmap;
 	int			indexes[FLEXIBLE_ARRAY_MEMBER];
 }			PartitionDispatchData;
 
@@ -298,7 +298,7 @@ ExecFindPartition(ModifyTableState *mtstate,
 	dispatch = pd[0];
 	while (true)
 	{
-		AttrNumber *map = dispatch->tupmap;
+		AttrMap    *map = dispatch->tupmap;
 		int			partidx = -1;
 
 		CHECK_FOR_INTERRUPTS();
@@ -511,7 +511,7 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 	Relation	firstResultRel = mtstate->resultRelInfo[0].ri_RelationDesc;
 	ResultRelInfo *leaf_part_rri;
 	MemoryContext oldcxt;
-	AttrNumber *part_attnos = NULL;
+	AttrMap    *part_attmap = NULL;
 	bool		found_whole_row;
 
 	oldcxt = MemoryContextSwitchTo(proute->memcxt);
@@ -584,14 +584,13 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 		/*
 		 * Convert Vars in it to contain this partition's attribute numbers.
 		 */
-		part_attnos =
-			convert_tuples_by_name_map(RelationGetDescr(partrel),
-									   RelationGetDescr(firstResultRel));
+		part_attmap =
+			build_attrmap_by_name(RelationGetDescr(partrel),
+								  RelationGetDescr(firstResultRel));
 		wcoList = (List *)
 			map_variable_attnos((Node *) wcoList,
 								firstVarno, 0,
-								part_attnos,
-								RelationGetDescr(firstResultRel)->natts,
+								part_attmap,
 								RelationGetForm(partrel)->reltype,
 								&found_whole_row);
 		/* We ignore the value of found_whole_row. */
@@ -642,15 +641,14 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 		/*
 		 * Convert Vars in it to contain this partition's attribute numbers.
 		 */
-		if (part_attnos == NULL)
-			part_attnos =
-				convert_tuples_by_name_map(RelationGetDescr(partrel),
-										   RelationGetDescr(firstResultRel));
+		if (part_attmap == NULL)
+			part_attmap =
+				build_attrmap_by_name(RelationGetDescr(partrel),
+									  RelationGetDescr(firstResultRel));
 		returningList = (List *)
 			map_variable_attnos((Node *) returningList,
 								firstVarno, 0,
-								part_attnos,
-								RelationGetDescr(firstResultRel)->natts,
+								part_attmap,
 								RelationGetForm(partrel)->reltype,
 								&found_whole_row);
 		/* We ignore the value of found_whole_row. */
@@ -785,23 +783,21 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 				 * target relation (firstVarno).
 				 */
 				onconflset = (List *) copyObject((Node *) node->onConflictSet);
-				if (part_attnos == NULL)
-					part_attnos =
-						convert_tuples_by_name_map(RelationGetDescr(partrel),
-												   RelationGetDescr(firstResultRel));
+				if (part_attmap == NULL)
+					part_attmap =
+						build_attrmap_by_name(RelationGetDescr(partrel),
+											  RelationGetDescr(firstResultRel));
 				onconflset = (List *)
 					map_variable_attnos((Node *) onconflset,
 										INNER_VAR, 0,
-										part_attnos,
-										RelationGetDescr(firstResultRel)->natts,
+										part_attmap,
 										RelationGetForm(partrel)->reltype,
 										&found_whole_row);
 				/* We ignore the value of found_whole_row. */
 				onconflset = (List *)
 					map_variable_attnos((Node *) onconflset,
 										firstVarno, 0,
-										part_attnos,
-										RelationGetDescr(firstResultRel)->natts,
+										part_attmap,
 										RelationGetForm(partrel)->reltype,
 										&found_whole_row);
 				/* We ignore the value of found_whole_row. */
@@ -835,16 +831,14 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 					clause = (List *)
 						map_variable_attnos((Node *) clause,
 											INNER_VAR, 0,
-											part_attnos,
-											RelationGetDescr(firstResultRel)->natts,
+											part_attmap,
 											RelationGetForm(partrel)->reltype,
 											&found_whole_row);
 					/* We ignore the value of found_whole_row. */
 					clause = (List *)
 						map_variable_attnos((Node *) clause,
 											firstVarno, 0,
-											part_attnos,
-											RelationGetDescr(firstResultRel)->natts,
+											part_attmap,
 											RelationGetForm(partrel)->reltype,
 											&found_whole_row);
 					/* We ignore the value of found_whole_row. */
@@ -1036,8 +1030,8 @@ ExecInitPartitionDispatchInfo(EState *estate,
 		 * tuple descriptor when computing its partition key for tuple
 		 * routing.
 		 */
-		pd->tupmap = convert_tuples_by_name_map_if_req(RelationGetDescr(parent_pd->reldesc),
-													   tupdesc);
+		pd->tupmap = build_attrmap_by_name_if_req(RelationGetDescr(parent_pd->reldesc),
+												  tupdesc);
 		pd->tupslot = pd->tupmap ?
 			MakeSingleTupleTableSlot(tupdesc, &TTSOpsVirtual) : NULL;
 	}
@@ -1434,15 +1428,16 @@ adjust_partition_tlist(List *tlist, TupleConversionMap *map)
 {
 	List	   *new_tlist = NIL;
 	TupleDesc	tupdesc = map->outdesc;
-	AttrNumber *attrMap = map->attrMap;
+	AttrMap    *attrMap = map->attrMap;
 	AttrNumber	attrno;
 
+	Assert(tupdesc->natts == attrMap->maplen);
 	for (attrno = 1; attrno <= tupdesc->natts; attrno++)
 	{
 		Form_pg_attribute att_tup = TupleDescAttr(tupdesc, attrno - 1);
 		TargetEntry *tle;
 
-		if (attrMap[attrno - 1] != InvalidAttrNumber)
+		if (attrMap->attnums[attrno - 1] != InvalidAttrNumber)
 		{
 			Assert(!att_tup->attisdropped);
 
@@ -1450,7 +1445,7 @@ adjust_partition_tlist(List *tlist, TupleConversionMap *map)
 			 * Use the corresponding entry from the parent's tlist, adjusting
 			 * the resno the match the partition's attno.
 			 */
-			tle = (TargetEntry *) list_nth(tlist, attrMap[attrno - 1] - 1);
+			tle = (TargetEntry *) list_nth(tlist, attrMap->attnums[attrno - 1] - 1);
 			tle->resno = attrno;
 		}
 		else
diff --git a/src/backend/jit/llvm/llvmjit_expr.c b/src/backend/jit/llvm/llvmjit_expr.c
index a9d362100a..ffd887c71a 100644
--- a/src/backend/jit/llvm/llvmjit_expr.c
+++ b/src/backend/jit/llvm/llvmjit_expr.c
@@ -20,7 +20,6 @@
 
 #include "access/htup_details.h"
 #include "access/nbtree.h"
-#include "access/tupconvert.h"
 #include "catalog/objectaccess.h"
 #include "catalog/pg_type.h"
 #include "executor/execExpr.h"
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index b761fdfd7d..45bb31ecf8 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -917,7 +917,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 	Relation	relation;
 	TupleDesc	tupleDesc;
 	TupleConstr *constr;
-	AttrNumber *attmap;
+	AttrMap    *attmap;
 	AclResult	aclresult;
 	char	   *comment;
 	ParseCallbackState pcbstate;
@@ -974,7 +974,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 	 * since dropped columns in the source table aren't copied, so the new
 	 * table can have different column numbers.
 	 */
-	attmap = (AttrNumber *) palloc0(sizeof(AttrNumber) * tupleDesc->natts);
+	attmap = make_attrmap(tupleDesc->natts);
 
 	/*
 	 * Insert the copied attributes into the cxt for the new table definition.
@@ -1020,7 +1020,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 		 */
 		cxt->columns = lappend(cxt->columns, def);
 
-		attmap[parent_attno - 1] = list_length(cxt->columns);
+		attmap->attnums[parent_attno - 1] = list_length(cxt->columns);
 
 		/*
 		 * Copy default, if present and it should be copied.  We have separate
@@ -1051,7 +1051,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 
 			def->cooked_default = map_variable_attnos(this_default,
 													  1, 0,
-													  attmap, tupleDesc->natts,
+													  attmap,
 													  InvalidOid, &found_whole_row);
 
 			/*
@@ -1134,7 +1134,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 
 			ccbin_node = map_variable_attnos(stringToNode(ccbin),
 											 1, 0,
-											 attmap, tupleDesc->natts,
+											 attmap,
 											 InvalidOid, &found_whole_row);
 
 			/*
@@ -1200,7 +1200,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
 			/* Build CREATE INDEX statement to recreate the parent_index */
 			index_stmt = generateClonedIndexStmt(cxt->relation,
 												 parent_index,
-												 attmap, tupleDesc->natts,
+												 attmap,
 												 NULL);
 
 			/* Copy comment on index, if requested */
@@ -1332,7 +1332,7 @@ transformOfType(CreateStmtContext *cxt, TypeName *ofTypename)
  */
 IndexStmt *
 generateClonedIndexStmt(RangeVar *heapRel, Relation source_idx,
-						const AttrNumber *attmap, int attmap_length,
+						const AttrMap *attmap,
 						Oid *constraintOid)
 {
 	Oid			source_relid = RelationGetRelid(source_idx);
@@ -1552,7 +1552,7 @@ generateClonedIndexStmt(RangeVar *heapRel, Relation source_idx,
 			/* Adjust Vars to match new table's column numbering */
 			indexkey = map_variable_attnos(indexkey,
 										   1, 0,
-										   attmap, attmap_length,
+										   attmap,
 										   InvalidOid, &found_whole_row);
 
 			/* As in transformTableLikeClause, reject whole-row variables */
@@ -1659,7 +1659,7 @@ generateClonedIndexStmt(RangeVar *heapRel, Relation source_idx,
 		/* Adjust Vars to match new table's column numbering */
 		pred_tree = map_variable_attnos(pred_tree,
 										1, 0,
-										attmap, attmap_length,
+										attmap,
 										InvalidOid, &found_whole_row);
 
 		/* As in transformTableLikeClause, reject whole-row variables */
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index b386f8460d..f38c5b3ea4 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -267,7 +267,7 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
 		 */
 		desc = RelationGetDescr(entry->localrel);
 		oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
-		entry->attrmap = palloc(desc->natts * sizeof(AttrNumber));
+		entry->attrmap = make_attrmap(desc->natts);
 		MemoryContextSwitchTo(oldctx);
 
 		found = 0;
@@ -278,14 +278,14 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
 
 			if (attr->attisdropped || attr->attgenerated)
 			{
-				entry->attrmap[i] = -1;
+				entry->attrmap->attnums[i] = -1;
 				continue;
 			}
 
 			attnum = logicalrep_rel_att_by_name(remoterel,
 												NameStr(attr->attname));
 
-			entry->attrmap[i] = attnum;
+			entry->attrmap->attnums[i] = attnum;
 			if (attnum >= 0)
 				found++;
 		}
@@ -340,8 +340,8 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
 
 			attnum = AttrNumberGetAttrOffset(attnum);
 
-			if (entry->attrmap[attnum] < 0 ||
-				!bms_is_member(entry->attrmap[attnum], remoterel->attkeys))
+			if (entry->attrmap->attnums[attnum] < 0 ||
+				!bms_is_member(entry->attrmap->attnums[attnum], remoterel->attkeys))
 			{
 				entry->updatable = false;
 				break;
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index ced0d191c2..b1d783383f 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -232,6 +232,7 @@ slot_fill_defaults(LogicalRepRelMapEntry *rel, EState *estate,
 	defmap = (int *) palloc(num_phys_attrs * sizeof(int));
 	defexprs = (ExprState **) palloc(num_phys_attrs * sizeof(ExprState *));
 
+	Assert(rel->attrmap->maplen == num_phys_attrs);
 	for (attnum = 0; attnum < num_phys_attrs; attnum++)
 	{
 		Expr	   *defexpr;
@@ -239,7 +240,7 @@ slot_fill_defaults(LogicalRepRelMapEntry *rel, EState *estate,
 		if (TupleDescAttr(desc, attnum)->attisdropped || TupleDescAttr(desc, attnum)->attgenerated)
 			continue;
 
-		if (rel->attrmap[attnum] >= 0)
+		if (rel->attrmap->attnums[attnum] >= 0)
 			continue;
 
 		defexpr = (Expr *) build_column_default(rel->localrel, attnum + 1);
@@ -321,10 +322,11 @@ slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel,
 	error_context_stack = &errcallback;
 
 	/* Call the "in" function for each non-dropped attribute */
+	Assert(natts == rel->attrmap->maplen);
 	for (i = 0; i < natts; i++)
 	{
 		Form_pg_attribute att = TupleDescAttr(slot->tts_tupleDescriptor, i);
-		int			remoteattnum = rel->attrmap[i];
+		int			remoteattnum = rel->attrmap->attnums[i];
 
 		if (!att->attisdropped && remoteattnum >= 0 &&
 			values[remoteattnum] != NULL)
@@ -405,10 +407,11 @@ slot_modify_cstrings(TupleTableSlot *slot, TupleTableSlot *srcslot,
 	error_context_stack = &errcallback;
 
 	/* Call the "in" function for each replaced attribute */
+	Assert(natts == rel->attrmap->maplen);
 	for (i = 0; i < natts; i++)
 	{
 		Form_pg_attribute att = TupleDescAttr(slot->tts_tupleDescriptor, i);
-		int			remoteattnum = rel->attrmap[i];
+		int			remoteattnum = rel->attrmap->attnums[i];
 
 		if (remoteattnum < 0)
 			continue;
diff --git a/src/backend/rewrite/rewriteManip.c b/src/backend/rewrite/rewriteManip.c
index 93508c2a87..12f7cadf3b 100644
--- a/src/backend/rewrite/rewriteManip.c
+++ b/src/backend/rewrite/rewriteManip.c
@@ -1221,8 +1221,7 @@ typedef struct
 {
 	int			target_varno;	/* RTE index to search for */
 	int			sublevels_up;	/* (current) nesting depth */
-	const AttrNumber *attno_map;	/* map array for user attnos */
-	int			map_length;		/* number of entries in attno_map[] */
+	const AttrMap *attno_map;	/* map array for user attnos */
 	Oid			to_rowtype;		/* change whole-row Vars to this type */
 	bool	   *found_whole_row;	/* output flag */
 } map_variable_attnos_context;
@@ -1249,11 +1248,11 @@ map_variable_attnos_mutator(Node *node,
 			if (attno > 0)
 			{
 				/* user-defined column, replace attno */
-				if (attno > context->map_length ||
-					context->attno_map[attno - 1] == 0)
+				if (attno > context->attno_map->maplen ||
+					context->attno_map->attnums[attno - 1] == 0)
 					elog(ERROR, "unexpected varattno %d in expression to be mapped",
 						 attno);
-				newvar->varattno = newvar->varoattno = context->attno_map[attno - 1];
+				newvar->varattno = newvar->varoattno = context->attno_map->attnums[attno - 1];
 			}
 			else if (attno == 0)
 			{
@@ -1350,7 +1349,7 @@ map_variable_attnos_mutator(Node *node,
 Node *
 map_variable_attnos(Node *node,
 					int target_varno, int sublevels_up,
-					const AttrNumber *attno_map, int map_length,
+					const AttrMap *attno_map,
 					Oid to_rowtype, bool *found_whole_row)
 {
 	map_variable_attnos_context context;
@@ -1358,7 +1357,6 @@ map_variable_attnos(Node *node,
 	context.target_varno = target_varno;
 	context.sublevels_up = sublevels_up;
 	context.attno_map = attno_map;
-	context.map_length = map_length;
 	context.to_rowtype = to_rowtype;
 	context.found_whole_row = found_whole_row;
 
diff --git a/src/include/access/attmap.h b/src/include/access/attmap.h
new file mode 100644
index 0000000000..57de8b6572
--- /dev/null
+++ b/src/include/access/attmap.h
@@ -0,0 +1,52 @@
+/*-------------------------------------------------------------------------
+ *
+ * attmap.h
+ *	  Definitions for PostgreSQL attribute mappings
+ *
+ *
+ * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/attmap.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef ATTMAP_H
+#define ATTMAP_H
+
+#include "access/attnum.h"
+#include "access/tupdesc.h"
+
+/*
+ * Attribute mapping structure
+ *
+ * This maps attribute numbers between a pair of relations, designated
+ * 'input' and 'output' (most typically inheritance parent and child
+ * relations), whose common columns may have different attribute numbers.
+ * Such difference may arise due to the columns being ordered differently
+ * in the two relations or the two relations having dropped columns at
+ * different positions.
+ *
+ * 'maplen' is set to the number of attributes of the 'output' relation,
+ * taking into account any of its dropped attributes, with the corresponding
+ * elements of the 'attnums' array set to 0.
+ */
+typedef struct AttrMap
+{
+	AttrNumber *attnums;
+	int			maplen;
+} AttrMap;
+
+extern AttrMap *make_attrmap(int maplen);
+extern void free_attrmap(AttrMap *map);
+
+/* Convertion routines to build mappings */
+extern AttrMap *build_attrmap_by_name(TupleDesc indesc,
+									  TupleDesc outdesc);
+extern AttrMap *build_attrmap_by_name_if_req(TupleDesc indesc,
+											 TupleDesc outdesc);
+extern AttrMap *build_attrmap_by_position(TupleDesc indesc,
+										  TupleDesc outdesc,
+										  const char *msg);
+
+#endif							/* ATTMAP_H */
diff --git a/src/include/access/tupconvert.h b/src/include/access/tupconvert.h
index 6d095f8e0d..5ed7fe6d39 100644
--- a/src/include/access/tupconvert.h
+++ b/src/include/access/tupconvert.h
@@ -14,6 +14,7 @@
 #ifndef TUPCONVERT_H
 #define TUPCONVERT_H
 
+#include "access/attmap.h"
 #include "access/htup.h"
 #include "access/tupdesc.h"
 #include "executor/tuptable.h"
@@ -23,7 +24,7 @@ typedef struct TupleConversionMap
 {
 	TupleDesc	indesc;			/* tupdesc for source rowtype */
 	TupleDesc	outdesc;		/* tupdesc for result rowtype */
-	AttrNumber *attrMap;		/* indexes of input fields, or 0 for null */
+	AttrMap    *attrMap;		/* indexes of input fields, or 0 for null */
 	Datum	   *invalues;		/* workspace for deconstructing source */
 	bool	   *inisnull;
 	Datum	   *outvalues;		/* workspace for constructing result */
@@ -38,14 +39,10 @@ extern TupleConversionMap *convert_tuples_by_position(TupleDesc indesc,
 extern TupleConversionMap *convert_tuples_by_name(TupleDesc indesc,
 												  TupleDesc outdesc);
 
-extern AttrNumber *convert_tuples_by_name_map(TupleDesc indesc,
-											  TupleDesc outdesc);
-extern AttrNumber *convert_tuples_by_name_map_if_req(TupleDesc indesc,
-													 TupleDesc outdesc);
-
 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 TupleTableSlot *execute_attr_map_slot(AttrMap *attrMap,
+											 TupleTableSlot *in_slot,
+											 TupleTableSlot *out_slot);
 
 extern void free_conversion_map(TupleConversionMap *map);
 
diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h
index 27d9e537d3..ea4ad1a2d1 100644
--- a/src/include/catalog/index.h
+++ b/src/include/catalog/index.h
@@ -111,7 +111,7 @@ extern IndexInfo *BuildDummyIndexInfo(Relation index);
 extern bool CompareIndexInfo(IndexInfo *info1, IndexInfo *info2,
 							 Oid *collations1, Oid *collations2,
 							 Oid *opfamilies1, Oid *opfamilies2,
-							 AttrNumber *attmap, int maplen);
+							 AttrMap *attmap);
 
 extern void BuildSpeculativeIndexInfo(Relation index, IndexInfo *ii);
 
diff --git a/src/include/parser/parse_utilcmd.h b/src/include/parser/parse_utilcmd.h
index 1348064ad0..08dd0ce4ca 100644
--- a/src/include/parser/parse_utilcmd.h
+++ b/src/include/parser/parse_utilcmd.h
@@ -16,6 +16,7 @@
 
 #include "parser/parse_node.h"
 
+typedef struct AttrMap AttrMap;
 
 extern List *transformCreateStmt(CreateStmt *stmt, const char *queryString);
 extern List *transformAlterTableStmt(Oid relid, AlterTableStmt *stmt,
@@ -29,7 +30,7 @@ extern PartitionBoundSpec *transformPartitionBound(ParseState *pstate, Relation
 												   PartitionBoundSpec *spec);
 extern IndexStmt *generateClonedIndexStmt(RangeVar *heapRel,
 										  Relation source_idx,
-										  const AttrNumber *attmap, int attmap_length,
+										  const AttrMap *attmap,
 										  Oid *constraintOid);
 
 #endif							/* PARSE_UTILCMD_H */
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index 2642a3f94e..950d99bfad 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -21,7 +21,7 @@ typedef struct LogicalRepRelMapEntry
 	/* Mapping to local relation, filled as needed. */
 	Oid			localreloid;	/* local relation id */
 	Relation	localrel;		/* relcache entry */
-	AttrNumber *attrmap;		/* map of local attributes to remote ones */
+	AttrMap    *attrmap;		/* map of local attributes to remote ones */
 	bool		updatable;		/* Can apply updates/deletes? */
 
 	/* Sync state. */
diff --git a/src/include/rewrite/rewriteManip.h b/src/include/rewrite/rewriteManip.h
index 8d8fd17e41..634cdc235d 100644
--- a/src/include/rewrite/rewriteManip.h
+++ b/src/include/rewrite/rewriteManip.h
@@ -17,6 +17,7 @@
 #include "nodes/parsenodes.h"
 
 
+typedef struct AttrMap AttrMap;
 typedef struct replace_rte_variables_context replace_rte_variables_context;
 
 typedef Node *(*replace_rte_variables_callback) (Var *var,
@@ -71,7 +72,7 @@ extern Node *replace_rte_variables_mutator(Node *node,
 
 extern Node *map_variable_attnos(Node *node,
 								 int target_varno, int sublevels_up,
-								 const AttrNumber *attno_map, int map_length,
+								 const AttrMap *attno_map,
 								 Oid to_rowtype, bool *found_whole_row);
 
 extern Node *ReplaceVarsFromTargetList(Node *node,
-- 
2.24.0