attmap.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * attmap.c
 *    Attribute mapping support.
 *
 * This file provides utility routines to build and manage attribute
 * mappings by comparing input and output TupleDescs.  Such mappings
 * are typically used by DDL operating on inheritance and partition trees
 * to do a conversion between rowtypes logically equivalent but with
 * columns in a different order, taking into account dropped columns.
 * They are also used by the tuple conversion routines in tupconvert.c.
 *
 * Portions Copyright (c) 1996-2021, 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 = TupleDescAttr(indesc, i);
        Form_pg_attribute outatt = TupleDescAttr(outdesc, i);

        /*
         * If the input column has a missing attribute, we need a conversion.
         */
        if (inatt->atthasmissing)
            return false;

        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.
         */
        if (attrMap->attnums[i] == 0 &&
            inatt->attisdropped &&
            inatt->attlen == outatt->attlen &&
            inatt->attalign == outatt->attalign)
            continue;

        return false;
    }

    return true;
}