Re: change_varattnos_of_a_node versus whole-row Vars

I wrote:
> change_varattnos_of_a_node(), which is used to adjust expressions
> referencing a parent relation or LIKE source relation to refer to the
> child relation, ignores whole-row Vars (those with attnum zero).
> ...
> My inclination, especially in the back branches, is to just throw error
> if we find a whole-row Var.
> ...
> There are a bunch of other things I don't like about
> change_varattnos_of_a_node, starting with the name, but those gripes are
> in the nature of cosmetics or inadequate error checking and wouldn't
> have any user-visible impact if changed.

Attached is a draft patch that adds error reports for whole-row Vars
and replaces change_varattnos_of_a_node with a more robust function.
The latter makes it kind of a large patch, but it's difficult to make
it a whole lot simpler unless we are willing to have the error messages
say only "cannot convert whole-row table reference", without any
indication of where the problem is.  That seems a tad unfriendly to me.

Comments?

            regards, tom lane

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 70753e33e430237a66991bac0740b8761c1b4e4e..d69809a2f866de511e7ce278b8dca9bf9db33d40 100644
*** a/src/backend/commands/tablecmds.c
--- b/src/backend/commands/tablecmds.c
***************
*** 68,73 ****
--- 68,74 ----
  #include "parser/parser.h"
  #include "rewrite/rewriteDefine.h"
  #include "rewrite/rewriteHandler.h"
+ #include "rewrite/rewriteManip.h"
  #include "storage/bufmgr.h"
  #include "storage/lmgr.h"
  #include "storage/lock.h"
*************** static void truncate_check_rel(Relation
*** 253,259 ****
  static List *MergeAttributes(List *schema, List *supers, char relpersistence,
                  List **supOids, List **supconstr, int *supOidCount);
  static bool MergeCheckConstraint(List *constraints, char *name, Node *expr);
- static bool change_varattnos_walker(Node *node, const AttrNumber *newattno);
  static void MergeAttributesIntoExisting(Relation child_rel, Relation parent_rel);
  static void MergeConstraintsIntoExisting(Relation child_rel, Relation parent_rel);
  static void StoreCatalogInheritance(Oid relationId, List *supers);
--- 254,259 ----
*************** MergeAttributes(List *schema, List *supe
*** 1496,1502 ****
           * parents after the first one, nor if we have dropped columns.)
           */
          newattno = (AttrNumber *)
!             palloc(tupleDesc->natts * sizeof(AttrNumber));

          for (parent_attno = 1; parent_attno <= tupleDesc->natts;
               parent_attno++)
--- 1496,1502 ----
           * parents after the first one, nor if we have dropped columns.)
           */
          newattno = (AttrNumber *)
!             palloc0(tupleDesc->natts * sizeof(AttrNumber));

          for (parent_attno = 1; parent_attno <= tupleDesc->natts;
               parent_attno++)
*************** MergeAttributes(List *schema, List *supe
*** 1510,1524 ****
               * Ignore dropped columns in the parent.
               */
              if (attribute->attisdropped)
!             {
!                 /*
!                  * change_varattnos_of_a_node asserts that this is greater
!                  * than zero, so if anything tries to use it, we should find
!                  * out.
!                  */
!                 newattno[parent_attno - 1] = 0;
!                 continue;
!             }

              /*
               * Does it conflict with some previously inherited column?
--- 1510,1516 ----
               * Ignore dropped columns in the parent.
               */
              if (attribute->attisdropped)
!                 continue;        /* leave newattno entry as zero */

              /*
               * Does it conflict with some previously inherited column?
*************** MergeAttributes(List *schema, List *supe
*** 1656,1669 ****
              {
                  char       *name = check[i].ccname;
                  Node       *expr;

                  /* ignore if the constraint is non-inheritable */
                  if (check[i].ccnoinherit)
                      continue;

!                 /* adjust varattnos of ccbin here */
!                 expr = stringToNode(check[i].ccbin);
!                 change_varattnos_of_a_node(expr, newattno);

                  /* check for duplicate */
                  if (!MergeCheckConstraint(constraints, name, expr))
--- 1648,1677 ----
              {
                  char       *name = check[i].ccname;
                  Node       *expr;
+                 bool        found_whole_row;

                  /* ignore if the constraint is non-inheritable */
                  if (check[i].ccnoinherit)
                      continue;

!                 /* Adjust Vars to match new table's column numbering */
!                 expr = map_variable_attnos(stringToNode(check[i].ccbin),
!                                            1, 0,
!                                            newattno, tupleDesc->natts,
!                                            &found_whole_row);
!
!                 /*
!                  * For the moment we have to reject whole-row variables.
!                  * We could convert them, if we knew the new table's rowtype
!                  * OID, but that hasn't been assigned yet.
!                  */
!                 if (found_whole_row)
!                     ereport(ERROR,
!                             (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
!                              errmsg("cannot convert whole-row table reference"),
!                              errdetail("Constraint \"%s\" contains a whole-row reference to table \"%s\".",
!                                        name,
!                                        RelationGetRelationName(relation))));

                  /* check for duplicate */
                  if (!MergeCheckConstraint(constraints, name, expr))
*************** MergeCheckConstraint(List *constraints,
*** 1867,1967 ****


  /*
-  * Replace varattno values in an expression tree according to the given
-  * map array, that is, varattno N is replaced by newattno[N-1].  It is
-  * caller's responsibility to ensure that the array is long enough to
-  * define values for all user varattnos present in the tree.  System column
-  * attnos remain unchanged.
-  *
-  * Note that the passed node tree is modified in-place!
-  */
- void
- change_varattnos_of_a_node(Node *node, const AttrNumber *newattno)
- {
-     /* no setup needed, so away we go */
-     (void) change_varattnos_walker(node, newattno);
- }
-
- static bool
- change_varattnos_walker(Node *node, const AttrNumber *newattno)
- {
-     if (node == NULL)
-         return false;
-     if (IsA(node, Var))
-     {
-         Var           *var = (Var *) node;
-
-         if (var->varlevelsup == 0 && var->varno == 1 &&
-             var->varattno > 0)
-         {
-             /*
-              * ??? the following may be a problem when the node is multiply
-              * referenced though stringToNode() doesn't create such a node
-              * currently.
-              */
-             Assert(newattno[var->varattno - 1] > 0);
-             var->varattno = var->varoattno = newattno[var->varattno - 1];
-         }
-         return false;
-     }
-     return expression_tree_walker(node, change_varattnos_walker,
-                                   (void *) newattno);
- }
-
- /*
-  * Generate a map for change_varattnos_of_a_node from old and new TupleDesc's,
-  * matching according to column name.
-  */
- AttrNumber *
- varattnos_map(TupleDesc olddesc, TupleDesc newdesc)
- {
-     AttrNumber *attmap;
-     int            i,
-                 j;
-
-     attmap = (AttrNumber *) palloc0(sizeof(AttrNumber) * olddesc->natts);
-     for (i = 1; i <= olddesc->natts; i++)
-     {
-         if (olddesc->attrs[i - 1]->attisdropped)
-             continue;            /* leave the entry as zero */
-
-         for (j = 1; j <= newdesc->natts; j++)
-         {
-             if (strcmp(NameStr(olddesc->attrs[i - 1]->attname),
-                        NameStr(newdesc->attrs[j - 1]->attname)) == 0)
-             {
-                 attmap[i - 1] = j;
-                 break;
-             }
-         }
-     }
-     return attmap;
- }
-
- /*
-  * Generate a map for change_varattnos_of_a_node from a TupleDesc and a list
-  * of ColumnDefs
-  */
- AttrNumber *
- varattnos_map_schema(TupleDesc old, List *schema)
- {
-     AttrNumber *attmap;
-     int            i;
-
-     attmap = (AttrNumber *) palloc0(sizeof(AttrNumber) * old->natts);
-     for (i = 1; i <= old->natts; i++)
-     {
-         if (old->attrs[i - 1]->attisdropped)
-             continue;            /* leave the entry as zero */
-
-         attmap[i - 1] = findAttrByName(NameStr(old->attrs[i - 1]->attname),
-                                        schema);
-     }
-     return attmap;
- }
-
-
- /*
   * StoreCatalogInheritance
   *        Updates the system catalogs with proper inheritance information.
   *
--- 1875,1880 ----
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index adc0d5b775e9754941d880722ea2fb7f4bf704fd..55a74eeaf448b825a091e50d8717082c89a71c5f 100644
*** a/src/backend/parser/parse_utilcmd.c
--- b/src/backend/parser/parse_utilcmd.c
*************** static void transformOfType(CreateStmtCo
*** 108,114 ****
                  TypeName *ofTypename);
  static char *chooseIndexName(const RangeVar *relation, IndexStmt *index_stmt);
  static IndexStmt *generateClonedIndexStmt(CreateStmtContext *cxt,
!                         Relation parent_index, AttrNumber *attmap);
  static List *get_collation(Oid collation, Oid actual_datatype);
  static List *get_opclass(Oid opclass, Oid actual_datatype);
  static void transformIndexConstraints(CreateStmtContext *cxt);
--- 108,115 ----
                  TypeName *ofTypename);
  static char *chooseIndexName(const RangeVar *relation, IndexStmt *index_stmt);
  static IndexStmt *generateClonedIndexStmt(CreateStmtContext *cxt,
!                         Relation source_idx,
!                         const AttrNumber *attmap, int attmap_length);
  static List *get_collation(Oid collation, Oid actual_datatype);
  static List *get_opclass(Oid opclass, Oid actual_datatype);
  static void transformIndexConstraints(CreateStmtContext *cxt);
*************** transformTableLikeClause(CreateStmtConte
*** 634,639 ****
--- 635,641 ----
      Relation    relation;
      TupleDesc    tupleDesc;
      TupleConstr *constr;
+     AttrNumber *attmap;
      AclResult    aclresult;
      char       *comment;
      ParseCallbackState pcbstate;
*************** transformTableLikeClause(CreateStmtConte
*** 677,682 ****
--- 679,691 ----
      constr = tupleDesc->constr;

      /*
+      * Initialize column number map for map_variable_attnos().  We need this
+      * 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);
+
+     /*
       * Insert the copied attributes into the cxt for the new table definition.
       */
      for (parent_attno = 1; parent_attno <= tupleDesc->natts;
*************** transformTableLikeClause(CreateStmtConte
*** 687,693 ****
          ColumnDef  *def;

          /*
!          * Ignore dropped columns in the parent.
           */
          if (attribute->attisdropped)
              continue;
--- 696,702 ----
          ColumnDef  *def;

          /*
!          * Ignore dropped columns in the parent.  attmap entry is left zero.
           */
          if (attribute->attisdropped)
              continue;
*************** transformTableLikeClause(CreateStmtConte
*** 718,723 ****
--- 727,734 ----
           */
          cxt->columns = lappend(cxt->columns, def);

+         attmap[parent_attno - 1] = list_length(cxt->columns);
+
          /*
           * Copy default, if present and the default has been requested
           */
*************** transformTableLikeClause(CreateStmtConte
*** 776,797 ****

      /*
       * Copy CHECK constraints if requested, being careful to adjust attribute
!      * numbers
       */
      if ((table_like_clause->options & CREATE_TABLE_LIKE_CONSTRAINTS) &&
          tupleDesc->constr)
      {
-         AttrNumber *attmap = varattnos_map_schema(tupleDesc, cxt->columns);
          int            ccnum;

          for (ccnum = 0; ccnum < tupleDesc->constr->num_check; ccnum++)
          {
              char       *ccname = tupleDesc->constr->check[ccnum].ccname;
              char       *ccbin = tupleDesc->constr->check[ccnum].ccbin;
-             Node       *ccbin_node = stringToNode(ccbin);
              Constraint *n = makeNode(Constraint);

!             change_varattnos_of_a_node(ccbin_node, attmap);

              n->contype = CONSTR_CHECK;
              n->location = -1;
--- 787,825 ----

      /*
       * Copy CHECK constraints if requested, being careful to adjust attribute
!      * numbers so they match the child.
       */
      if ((table_like_clause->options & CREATE_TABLE_LIKE_CONSTRAINTS) &&
          tupleDesc->constr)
      {
          int            ccnum;

          for (ccnum = 0; ccnum < tupleDesc->constr->num_check; ccnum++)
          {
              char       *ccname = tupleDesc->constr->check[ccnum].ccname;
              char       *ccbin = tupleDesc->constr->check[ccnum].ccbin;
              Constraint *n = makeNode(Constraint);
+             Node       *ccbin_node;
+             bool        found_whole_row;

!             ccbin_node = map_variable_attnos(stringToNode(ccbin),
!                                              1, 0,
!                                              attmap, tupleDesc->natts,
!                                              &found_whole_row);
!
!             /*
!              * We reject whole-row variables because the whole point of LIKE
!              * is that the new table's rowtype might later diverge from the
!              * parent's.  So, while translation might be possible right now,
!              * it wouldn't be possible to guarantee it would work in future.
!              */
!             if (found_whole_row)
!                 ereport(ERROR,
!                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
!                          errmsg("cannot convert whole-row table reference"),
!                          errdetail("Constraint \"%s\" contains a whole-row reference to table \"%s\".",
!                                    ccname,
!                                    RelationGetRelationName(relation))));

              n->contype = CONSTR_CHECK;
              n->location = -1;
*************** transformTableLikeClause(CreateStmtConte
*** 827,833 ****
      if ((table_like_clause->options & CREATE_TABLE_LIKE_INDEXES) &&
          relation->rd_rel->relhasindex)
      {
-         AttrNumber *attmap = varattnos_map_schema(tupleDesc, cxt->columns);
          List       *parent_indexes;
          ListCell   *l;

--- 855,860 ----
*************** transformTableLikeClause(CreateStmtConte
*** 842,848 ****
              parent_index = index_open(parent_index_oid, AccessShareLock);

              /* Build CREATE INDEX statement to recreate the parent_index */
!             index_stmt = generateClonedIndexStmt(cxt, parent_index, attmap);

              /* Copy comment on index */
              if (table_like_clause->options & CREATE_TABLE_LIKE_COMMENTS)
--- 869,876 ----
              parent_index = index_open(parent_index_oid, AccessShareLock);

              /* Build CREATE INDEX statement to recreate the parent_index */
!             index_stmt = generateClonedIndexStmt(cxt, parent_index,
!                                                  attmap, tupleDesc->natts);

              /* Copy comment on index */
              if (table_like_clause->options & CREATE_TABLE_LIKE_COMMENTS)
*************** chooseIndexName(const RangeVar *relation
*** 961,967 ****
   */
  static IndexStmt *
  generateClonedIndexStmt(CreateStmtContext *cxt, Relation source_idx,
!                         AttrNumber *attmap)
  {
      Oid            source_relid = RelationGetRelid(source_idx);
      Form_pg_attribute *attrs = RelationGetDescr(source_idx)->attrs;
--- 989,995 ----
   */
  static IndexStmt *
  generateClonedIndexStmt(CreateStmtContext *cxt, Relation source_idx,
!                         const AttrNumber *attmap, int attmap_length)
  {
      Oid            source_relid = RelationGetRelid(source_idx);
      Form_pg_attribute *attrs = RelationGetDescr(source_idx)->attrs;
*************** generateClonedIndexStmt(CreateStmtContex
*** 1149,1162 ****
          {
              /* Expressional index */
              Node       *indexkey;

              if (indexpr_item == NULL)
                  elog(ERROR, "too few entries in indexprs list");
              indexkey = (Node *) lfirst(indexpr_item);
              indexpr_item = lnext(indexpr_item);

!             /* OK to modify indexkey since we are working on a private copy */
!             change_varattnos_of_a_node(indexkey, attmap);

              iparam->name = NULL;
              iparam->expr = indexkey;
--- 1177,1202 ----
          {
              /* Expressional index */
              Node       *indexkey;
+             bool        found_whole_row;

              if (indexpr_item == NULL)
                  elog(ERROR, "too few entries in indexprs list");
              indexkey = (Node *) lfirst(indexpr_item);
              indexpr_item = lnext(indexpr_item);

!             /* Adjust Vars to match new table's column numbering */
!             indexkey = map_variable_attnos(indexkey,
!                                            1, 0,
!                                            attmap, attmap_length,
!                                            &found_whole_row);
!
!             /* As in transformTableLikeClause, reject whole-row variables */
!             if (found_whole_row)
!                 ereport(ERROR,
!                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
!                          errmsg("cannot convert whole-row table reference"),
!                          errdetail("Index \"%s\" contains a whole-row table reference.",
!                                    RelationGetRelationName(source_idx))));

              iparam->name = NULL;
              iparam->expr = indexkey;
*************** generateClonedIndexStmt(CreateStmtContex
*** 1213,1224 ****
      if (!isnull)
      {
          char       *pred_str;

          /* Convert text string to node tree */
          pred_str = TextDatumGetCString(datum);
!         index->whereClause = (Node *) stringToNode(pred_str);
!         /* Adjust attribute numbers */
!         change_varattnos_of_a_node(index->whereClause, attmap);
      }

      /* Clean up */
--- 1253,1280 ----
      if (!isnull)
      {
          char       *pred_str;
+         Node       *pred_tree;
+         bool        found_whole_row;

          /* Convert text string to node tree */
          pred_str = TextDatumGetCString(datum);
!         pred_tree = (Node *) stringToNode(pred_str);
!
!         /* Adjust Vars to match new table's column numbering */
!         pred_tree = map_variable_attnos(pred_tree,
!                                         1, 0,
!                                         attmap, attmap_length,
!                                         &found_whole_row);
!
!         /* As in transformTableLikeClause, reject whole-row variables */
!         if (found_whole_row)
!             ereport(ERROR,
!                     (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
!                      errmsg("cannot convert whole-row table reference"),
!                      errdetail("Index \"%s\" contains a whole-row table reference.",
!                                RelationGetRelationName(source_idx))));
!
!         index->whereClause = pred_tree;
      }

      /* Clean up */
diff --git a/src/backend/rewrite/rewriteManip.c b/src/backend/rewrite/rewriteManip.c
index ec6b5bf5a924a8655227be1eeaffeefa88927611..9c778efd1c39849726c775aab55e3bdf3d97487f 100644
*** a/src/backend/rewrite/rewriteManip.c
--- b/src/backend/rewrite/rewriteManip.c
*************** replace_rte_variables_mutator(Node *node
*** 1218,1223 ****
--- 1218,1336 ----


  /*
+  * map_variable_attnos() finds all user-column Vars in an expression tree
+  * that reference a particular RTE, and adjusts their varattnos according
+  * to the given mapping array (varattno n is replaced by attno_map[n-1]).
+  * Vars for system columns are not modified.
+  *
+  * A zero in the mapping array represents a dropped column, which should not
+  * appear in the expression.
+  *
+  * If the expression tree contains a whole-row Var for the target RTE,
+  * the Var is not changed but *found_whole_row is returned as TRUE.
+  * For most callers this is an error condition, but we leave it to the caller
+  * to report the error so that useful context can be provided.  (In some
+  * usages it would be appropriate to modify the Var's vartype and insert a
+  * ConvertRowtypeExpr node to map back to the original vartype.  We might
+  * someday extend this function's API to support that.  For now, the only
+  * concession to that future need is that this function is a tree mutator
+  * not just a walker.)
+  *
+  * This could be built using replace_rte_variables and a callback function,
+  * but since we don't ever need to insert sublinks, replace_rte_variables is
+  * overly complicated.
+  */
+
+ 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[] */
+     bool       *found_whole_row;    /* output flag */
+ } map_variable_attnos_context;
+
+ static Node *
+ map_variable_attnos_mutator(Node *node,
+                             map_variable_attnos_context *context)
+ {
+     if (node == NULL)
+         return NULL;
+     if (IsA(node, Var))
+     {
+         Var           *var = (Var *) node;
+
+         if (var->varno == context->target_varno &&
+             var->varlevelsup == context->sublevels_up)
+         {
+             /* Found a matching variable, make the substitution */
+             Var       *newvar = (Var *) palloc(sizeof(Var));
+             int        attno = var->varattno;
+
+             *newvar = *var;
+             if (attno > 0)
+             {
+                 /* user-defined column, replace attno */
+                 if (attno > context->map_length ||
+                     context->attno_map[attno - 1] == 0)
+                     elog(ERROR, "unexpected varattno %d in expression to be mapped",
+                          attno);
+                 newvar->varattno = newvar->varoattno = context->attno_map[attno - 1];
+             }
+             else if (attno == 0)
+             {
+                 /* whole-row variable, warn caller */
+                 *(context->found_whole_row) = true;
+             }
+             return (Node *) newvar;
+         }
+         /* otherwise fall through to copy the var normally */
+     }
+     else if (IsA(node, Query))
+     {
+         /* Recurse into RTE subquery or not-yet-planned sublink subquery */
+         Query       *newnode;
+
+         context->sublevels_up++;
+         newnode = query_tree_mutator((Query *) node,
+                                      map_variable_attnos_mutator,
+                                      (void *) context,
+                                      0);
+         context->sublevels_up--;
+         return (Node *) newnode;
+     }
+     return expression_tree_mutator(node, map_variable_attnos_mutator,
+                                    (void *) context);
+ }
+
+ Node *
+ map_variable_attnos(Node *node,
+                     int target_varno, int sublevels_up,
+                     const AttrNumber *attno_map, int map_length,
+                     bool *found_whole_row)
+ {
+     map_variable_attnos_context context;
+
+     context.target_varno = target_varno;
+     context.sublevels_up = sublevels_up;
+     context.attno_map = attno_map;
+     context.map_length = map_length;
+     context.found_whole_row = found_whole_row;
+
+     *found_whole_row = false;
+
+     /*
+      * Must be prepared to start with a Query or a bare expression tree; if
+      * it's a Query, we don't want to increment sublevels_up.
+      */
+     return query_or_expression_tree_mutator(node,
+                                             map_variable_attnos_mutator,
+                                             (void *) &context,
+                                             0);
+ }
+
+
+ /*
   * ResolveNew - replace Vars with corresponding items from a targetlist
   *
   * Vars matching target_varno and sublevels_up are replaced by the
diff --git a/src/include/commands/tablecmds.h b/src/include/commands/tablecmds.h
index 9ceb086f681420be255ae8790988fb074b6495ff..15d4713cec16ff9e2c4366deedf6aca807516cf5 100644
*** a/src/include/commands/tablecmds.h
--- b/src/include/commands/tablecmds.h
*************** extern void find_composite_type_dependen
*** 61,70 ****

  extern void check_of_type(HeapTuple typetuple);

- extern AttrNumber *varattnos_map(TupleDesc olddesc, TupleDesc newdesc);
- extern AttrNumber *varattnos_map_schema(TupleDesc old, List *schema);
- extern void change_varattnos_of_a_node(Node *node, const AttrNumber *newattno);
-
  extern void register_on_commit_action(Oid relid, OnCommitAction action);
  extern void remove_on_commit_action(Oid relid);

--- 61,66 ----
diff --git a/src/include/rewrite/rewriteManip.h b/src/include/rewrite/rewriteManip.h
index 7226187849232caf9a6c1dc30f2334c7e2656283..6f57b37b81443ba1a5b7a450b2cb005b99831187 100644
*** a/src/include/rewrite/rewriteManip.h
--- b/src/include/rewrite/rewriteManip.h
*************** extern Node *replace_rte_variables(Node
*** 65,70 ****
--- 65,75 ----
  extern Node *replace_rte_variables_mutator(Node *node,
                                replace_rte_variables_context *context);

+ extern Node *map_variable_attnos(Node *node,
+                     int target_varno, int sublevels_up,
+                     const AttrNumber *attno_map, int map_length,
+                     bool *found_whole_row);
+
  extern Node *ResolveNew(Node *node, int target_varno, int sublevels_up,
             RangeTblEntry *target_rte,
             List *targetlist, int event, int update_varno,