Re: PL/PGSQL: Dynamic Record Introspection - Mailing list pgsql-patches

OK, patch reverted, and attached. Would the author please revise?
Thanks.

It seems like a cool feature.

---------------------------------------------------------------------------

Tom Lane wrote:
> Bruce Momjian <pgman@candle.pha.pa.us> writes:
> > Patch applied.  Thanks.
>
> I wish to object to this patch; it's poorly designed, poorly coded, and
> badly documented.  The philosophy seems to be "I want this feature
> and I don't care what I have to do to the semantics and performance
> of plpgsql to get it".  In particular I don't like the bits that go
>
>                  /* Do not allow typeids to become "narrowed" by InvalidOids
>                  causing specialized typeids from the tuple restricting the destination */
>
> The incoherence of the comment is a good guide to how poorly thought out
> the code is.  These bits are positioned so that they change the language
> semantics even when not using a dynamic field reference; what's more,
> they don't make any sense when you *are* using a dynamic field
> reference, because you need to keep the actual field type not try
> (probably vainly) to cast it to whatever the previous field's type was.
>
> I also dislike the loop added to exec_eval_cleanup(), which will impose
> a significant performance penalty on every single expression evaluation
> done by any plpgsql function, whether it's ever heard of dynamic field
> references or not.  I doubt it's even necessary; I think the author
> doesn't understand plpgsql's memory allocation strategy very well.
>
> Lastly, and this is maybe a judgement call, I find the changes to
> PLpgSQL_recfield fairly ugly.  It'd be better to make a separate
> datum type PLpgSQL_dynamic_recfield or some such.
>
> This needs to be bounced back for rework.  It looks like a first draft
> that should have been rewritten once the author had learned enough to
> make it sort-of-work.
>
>             regards, tom lane
>

--
  Bruce Momjian   http://candle.pha.pa.us
  EnterpriseDB    http://www.enterprisedb.com

  + If your life is a hard drive, Christ can be your backup. +
*** ./doc/src/sgml/plpgsql.sgml.orig    Fri Mar 10 20:10:48 2006
--- ./doc/src/sgml/plpgsql.sgml    Mon May  8 23:40:12 2006
***************
*** 865,870 ****
--- 865,919 ----
     </para>

     <para>
+     To obtain the values of the fields the record is made up of,
+     the record variable can be qualified with the column or field
+     name. This can be done either by literally using the column name
+     or the column name for indexing the record can be taken out of a scalar
+     variable. The syntax for this notation is Record_variable.(IndexVariable).
+     To get information about the column field names of the record,
+     a special expression exists that returns all column names as an array:
+     RecordVariable.(*) .
+     Thus, the RECORD can be viewed
+     as an associative array that allows for introspection of it's contents.
+     This feature is especially useful for writing generic triggers that
+     operate on records with unknown structure.
+     Here is an example procedure that shows column names and values
+     of the predefined record NEW in a trigger procedure:
+ <programlisting>
+
+ CREATE OR REPLACE FUNCTION show_associative_records() RETURNS TRIGGER AS $$
+     DECLARE
+         colname        TEXT;
+         colcontent    TEXT;
+         colnames    TEXT[];
+         coln        INT4;
+         coli        INT4;
+     BEGIN
+ -- obtain an array with all field names of the record
+         colnames := NEW.(*);
+         RAISE NOTICE 'All column names of test record: %', colnames;
+ -- show field names and contents of record
+         coli := 1;
+         coln := array_upper(colnames,1);
+         RAISE NOTICE 'Number of columns in NEW: %', coln;
+         FOR coli IN 1 .. coln LOOP
+             colname := colnames[coli];
+             colcontent := NEW.(colname);
+             RAISE NOTICE 'column % of NEW: %', quote_ident(colname), quote_literal(colcontent);
+         END LOOP;
+ -- Do it with a fixed field name:
+ -- will have to know the column name
+         RAISE NOTICE 'column someint of NEW: %', quote_literal(NEW.someint);
+         RETURN NULL;
+     END;
+ $$ LANGUAGE plpgsql;
+ --CREATE TABLE test_records (someint INT8, somestring TEXT);
+ --CREATE TRIGGER tr_test_record BEFORE INSERT ON test_records FOR EACH ROW EXECUTE PROCEDURE
show_associative_records();
+
+ </programlisting>
+    </para>
+
+    <para>
      Note that <literal>RECORD</> is not a true data type, only a placeholder.
      One should also realize that when a <application>PL/pgSQL</application>
      function is declared to return type <type>record</>, this is not quite the
*** ./src/pl/plpgsql/src/pl_comp.c.orig    Tue Mar 14 23:48:23 2006
--- ./src/pl/plpgsql/src/pl_comp.c    Mon May  8 22:49:50 2006
***************
*** 870,876 ****

                  new = palloc(sizeof(PLpgSQL_recfield));
                  new->dtype = PLPGSQL_DTYPE_RECFIELD;
!                 new->fieldname = pstrdup(cp[1]);
                  new->recparentno = ns->itemno;

                  plpgsql_adddatum((PLpgSQL_datum *) new);
--- 870,877 ----

                  new = palloc(sizeof(PLpgSQL_recfield));
                  new->dtype = PLPGSQL_DTYPE_RECFIELD;
!                 new->fieldindex.fieldname = pstrdup(cp[1]);
!                 new->fieldindex_flag = RECFIELD_USE_FIELDNAME;
                  new->recparentno = ns->itemno;

                  plpgsql_adddatum((PLpgSQL_datum *) new);
***************
*** 976,982 ****

                  new = palloc(sizeof(PLpgSQL_recfield));
                  new->dtype = PLPGSQL_DTYPE_RECFIELD;
!                 new->fieldname = pstrdup(cp[2]);
                  new->recparentno = ns->itemno;

                  plpgsql_adddatum((PLpgSQL_datum *) new);
--- 977,984 ----

                  new = palloc(sizeof(PLpgSQL_recfield));
                  new->dtype = PLPGSQL_DTYPE_RECFIELD;
!                 new->fieldindex.fieldname = pstrdup(cp[2]);
!                 new->fieldindex_flag = RECFIELD_USE_FIELDNAME;
                  new->recparentno = ns->itemno;

                  plpgsql_adddatum((PLpgSQL_datum *) new);
***************
*** 1423,1428 ****
--- 1425,1556 ----
      MemoryContextSwitchTo(oldCxt);
      return T_DTYPE;
  }
+
+ /* ----------
+  * plpgsql_parse_recindex
+  * lookup associative index into record
+  * ----------
+  */
+ int
+ plpgsql_parse_recindex(char *word)
+ {
+     PLpgSQL_nsitem *ns1, *ns2;
+     char        *cp[2];
+     int        ret = T_ERROR;
+     char        *fieldvar;
+     int        fl;
+
+     /* Do case conversion and word separation */
+     plpgsql_convert_ident(word, cp, 2);
+     Assert(cp[1] != NULL);
+
+     /* cleanup the "(identifier)" string to "identifier" */
+     fieldvar = cp[1];
+     Assert(*fieldvar == '(');
+     ++fieldvar;    /* get rid of ( */
+
+     fl = strlen(fieldvar);
+     Assert(fieldvar[fl-1] == ')');
+     fieldvar[fl-1] = 0; /* get rid of ) */
+
+     /*
+      * Lookup the first word
+      */
+     ns1 = plpgsql_ns_lookup(cp[0], NULL);
+     if ( ns1 == NULL )
+     {
+         pfree(cp[0]);
+         pfree(cp[1]);
+         return T_ERROR;
+     }
+
+     ns2 = plpgsql_ns_lookup(fieldvar, NULL);
+     pfree(cp[0]);
+     pfree(cp[1]);
+     if ( ns2 == NULL )    /* name lookup failed */
+         return T_ERROR;
+
+     switch (ns1->itemtype)
+     {
+         case PLPGSQL_NSTYPE_REC:
+             {
+                 /*
+                  * First word is a record name, so second word must be an
+                  * variable holding the field name in this record.
+                  */
+                 if ( ns2->itemtype == PLPGSQL_NSTYPE_VAR ) {
+                     PLpgSQL_recfield *new;
+
+                     new = palloc(sizeof(PLpgSQL_recfield));
+                     new->dtype = PLPGSQL_DTYPE_RECFIELD;
+                     new->fieldindex.indexvar_no = ns2->itemno;
+                     new->fieldindex_flag = RECFIELD_USE_INDEX_VAR;
+                     new->recparentno = ns1->itemno;
+
+                     plpgsql_adddatum((PLpgSQL_datum *) new);
+
+                     plpgsql_yylval.scalar = (PLpgSQL_datum *) new;
+                     ret =  T_SCALAR;
+                 }
+                 break;
+             }
+         default:
+             break;
+     }
+     return ret;
+ }
+
+
+ /* ----------
+  * plpgsql_parse_recfieldnames
+  * create fieldnames of a record
+  * ----------
+  */
+ int
+ plpgsql_parse_recfieldnames(char *word)
+ {
+     PLpgSQL_nsitem    *ns1;
+     char        *cp[2];
+     int        ret = T_ERROR;
+
+     /* Do case conversion and word separation */
+     plpgsql_convert_ident(word, cp, 2);
+
+     /*
+      * Lookup the first word
+      */
+     ns1 = plpgsql_ns_lookup(cp[0], NULL);
+     if ( ns1 == NULL )
+     {
+         pfree(cp[0]);
+         pfree(cp[1]);
+         return T_ERROR;
+     }
+
+     pfree(cp[0]);
+     pfree(cp[1]);
+
+     switch (ns1->itemtype)
+     {
+         case PLPGSQL_NSTYPE_REC:
+             {
+                 PLpgSQL_recfieldproperties *new;
+
+                 new = palloc(sizeof(PLpgSQL_recfieldproperties));
+                 new->dtype = PLPGSQL_DTYPE_RECFIELDNAMES;
+                 new->recparentno = ns1->itemno;
+                 new->save_fieldnames = NULL;
+                 plpgsql_adddatum((PLpgSQL_datum *) new);
+                 plpgsql_yylval.scalar = (PLpgSQL_datum *) new;
+                 ret =  T_SCALAR;    /* ??? */
+                 break;
+             }
+         default:
+             break;
+     }
+     return ret;
+ }
+

  /*
   * plpgsql_build_variable - build a datum-array entry of a given
*** ./src/pl/plpgsql/src/pl_exec.c.orig    Sat Apr 22 03:26:01 2006
--- ./src/pl/plpgsql/src/pl_exec.c    Mon May  8 23:53:25 2006
***************
*** 726,732 ****
          case PLPGSQL_DTYPE_RECFIELD:
          case PLPGSQL_DTYPE_ARRAYELEM:
          case PLPGSQL_DTYPE_TRIGARG:
!
              /*
               * These datum records are read-only at runtime, so no need to
               * copy them
--- 726,732 ----
          case PLPGSQL_DTYPE_RECFIELD:
          case PLPGSQL_DTYPE_ARRAYELEM:
          case PLPGSQL_DTYPE_TRIGARG:
!         case PLPGSQL_DTYPE_RECFIELDNAMES:
              /*
               * These datum records are read-only at runtime, so no need to
               * copy them
***************
*** 836,841 ****
--- 836,842 ----

              case PLPGSQL_DTYPE_RECFIELD:
              case PLPGSQL_DTYPE_ARRAYELEM:
+             case PLPGSQL_DTYPE_RECFIELDNAMES:
                  break;

              default:
***************
*** 2164,2169 ****
--- 2165,2172 ----
  static void
  exec_eval_cleanup(PLpgSQL_execstate *estate)
  {
+     int        i;
+     ArrayType    *a;
      /* Clear result of a full SPI_execute */
      if (estate->eval_tuptable != NULL)
          SPI_freetuptable(estate->eval_tuptable);
***************
*** 2172,2177 ****
--- 2175,2188 ----
      /* Clear result of exec_eval_simple_expr (but keep the econtext) */
      if (estate->eval_econtext != NULL)
          ResetExprContext(estate->eval_econtext);
+     for ( i = 0; i < estate->ndatums; ++i ) {
+         if ( estate->datums[i]->dtype == PLPGSQL_DTYPE_RECFIELDNAMES ) {
+             a = ((PLpgSQL_recfieldproperties *)(estate->datums[i]))->save_fieldnames;
+             if ( a )
+                 pfree(a);
+             ((PLpgSQL_recfieldproperties *)(estate->datums[i]))->save_fieldnames = NULL;
+         }
+     }
  }


***************
*** 3141,3147 ****
                   */
                  PLpgSQL_recfield *recfield = (PLpgSQL_recfield *) target;
                  PLpgSQL_rec *rec;
!                 int            fno;
                  HeapTuple    newtup;
                  int            natts;
                  int            i;
--- 3152,3158 ----
                   */
                  PLpgSQL_recfield *recfield = (PLpgSQL_recfield *) target;
                  PLpgSQL_rec *rec;
!                 int            fno = 0;
                  HeapTuple    newtup;
                  int            natts;
                  int            i;
***************
*** 3170,3181 ****
                   * Get the number of the records field to change and the
                   * number of attributes in the tuple.
                   */
!                 fno = SPI_fnumber(rec->tupdesc, recfield->fieldname);
!                 if (fno == SPI_ERROR_NOATTRIBUTE)
                      ereport(ERROR,
!                             (errcode(ERRCODE_UNDEFINED_COLUMN),
!                              errmsg("record \"%s\" has no field \"%s\"",
!                                     rec->refname, recfield->fieldname)));
                  fno--;
                  natts = rec->tupdesc->natts;

--- 3181,3215 ----
                   * Get the number of the records field to change and the
                   * number of attributes in the tuple.
                   */
!                 if ( recfield->fieldindex_flag == RECFIELD_USE_FIELDNAME ) {
!                     fno = SPI_fnumber(rec->tupdesc, recfield->fieldindex.fieldname);
!                     if (fno == SPI_ERROR_NOATTRIBUTE)
!                         ereport(ERROR,
!                                 (errcode(ERRCODE_UNDEFINED_COLUMN),
!                                  errmsg("record \"%s\" has no field \"%s\"",
!                                         rec->refname, recfield->fieldindex.fieldname)));
!                 }
!                 else if ( recfield->fieldindex_flag == RECFIELD_USE_INDEX_VAR ) {
!                     PLpgSQL_var * idxvar = (PLpgSQL_var *) (estate->datums[recfield->fieldindex.indexvar_no]);
!                     char * fname = convert_value_to_string(idxvar->value, idxvar->datatype->typoid);
!                     if ( fname == NULL )
!                         ereport(ERROR,
!                                 (errcode(ERRCODE_UNDEFINED_COLUMN),
!                                 errmsg("record \"%s\": cannot evaluate variable to record index string",
!                                         rec->refname)));
!                     fno = SPI_fnumber(rec->tupdesc, fname);
!                     pfree(fname);
!                     if (fno == SPI_ERROR_NOATTRIBUTE)
!                         ereport(ERROR,
!                                 (errcode(ERRCODE_UNDEFINED_COLUMN),
!                                  errmsg("record \"%s\" has no field \"%s\"",
!                                         rec->refname, fname)));
!                 }
!                 else
                      ereport(ERROR,
!                         (errcode(ERRCODE_UNDEFINED_COLUMN),
!                         errmsg("record \"%s\": internal error",
!                                     rec->refname)));
                  fno--;
                  natts = rec->tupdesc->natts;

***************
*** 3495,3501 ****
              {
                  PLpgSQL_recfield *recfield = (PLpgSQL_recfield *) datum;
                  PLpgSQL_rec *rec;
!                 int            fno;

                  rec = (PLpgSQL_rec *) (estate->datums[recfield->recparentno]);
                  if (!HeapTupleIsValid(rec->tup))
--- 3529,3535 ----
              {
                  PLpgSQL_recfield *recfield = (PLpgSQL_recfield *) datum;
                  PLpgSQL_rec *rec;
!                 int            fno = 0;

                  rec = (PLpgSQL_rec *) (estate->datums[recfield->recparentno]);
                  if (!HeapTupleIsValid(rec->tup))
***************
*** 3504,3525 ****
                             errmsg("record \"%s\" is not assigned yet",
                                    rec->refname),
                             errdetail("The tuple structure of a not-yet-assigned record is indeterminate.")));
!                 fno = SPI_fnumber(rec->tupdesc, recfield->fieldname);
!                 if (fno == SPI_ERROR_NOATTRIBUTE)
!                     ereport(ERROR,
!                             (errcode(ERRCODE_UNDEFINED_COLUMN),
!                              errmsg("record \"%s\" has no field \"%s\"",
!                                     rec->refname, recfield->fieldname)));
!                 *typeid = SPI_gettypeid(rec->tupdesc, fno);
!                 *value = SPI_getbinval(rec->tup, rec->tupdesc, fno, isnull);
!                 if (expectedtypeid != InvalidOid && expectedtypeid != *typeid)
!                     ereport(ERROR,
!                             (errcode(ERRCODE_DATATYPE_MISMATCH),
!                              errmsg("type of \"%s.%s\" does not match that when preparing the plan",
!                                     rec->refname, recfield->fieldname)));
!                 break;
!             }
!
          case PLPGSQL_DTYPE_TRIGARG:
              {
                  PLpgSQL_trigarg *trigarg = (PLpgSQL_trigarg *) datum;
--- 3538,3662 ----
                             errmsg("record \"%s\" is not assigned yet",
                                    rec->refname),
                             errdetail("The tuple structure of a not-yet-assigned record is indeterminate.")));
!                  if ( recfield->fieldindex_flag == RECFIELD_USE_FIELDNAME ) {
!                      fno = SPI_fnumber(rec->tupdesc, recfield->fieldindex.fieldname);
!                      if (fno == SPI_ERROR_NOATTRIBUTE)
!                          ereport(ERROR,
!                                  (errcode(ERRCODE_UNDEFINED_COLUMN),
!                                   errmsg("record \"%s\" has no field \"%s\"",
!                                          rec->refname, recfield->fieldindex.fieldname)));
!                  }
!                  else if ( recfield->fieldindex_flag == RECFIELD_USE_INDEX_VAR ) {
!                      PLpgSQL_var * idxvar = (PLpgSQL_var *) (estate->datums[recfield->fieldindex.indexvar_no]);
!                      char * fname = convert_value_to_string(idxvar->value, idxvar->datatype->typoid);
!                      if ( fname == NULL )
!                          ereport(ERROR,
!                                  (errcode(ERRCODE_UNDEFINED_COLUMN),
!                                  errmsg("record \"%s\": cannot evaluate variable to record index string",
!                                          rec->refname)));
!                      fno = SPI_fnumber(rec->tupdesc, fname);
!                      pfree(fname);
!                      if (fno == SPI_ERROR_NOATTRIBUTE)
!                          ereport(ERROR,
!                                  (errcode(ERRCODE_UNDEFINED_COLUMN),
!                                   errmsg("record \"%s\" has no field \"%s\"",
!                                          rec->refname, fname)));
!                  }
!                  else
!                      ereport(ERROR,
!                          (errcode(ERRCODE_UNDEFINED_COLUMN),
!                          errmsg("record \"%s\": internal error",
!                                  rec->refname)));
!
!                  /* Do not allow typeids to become "narrowed" by InvalidOids
!                  causing specialized typeids from the tuple restricting the destination */
!                  if ( expectedtypeid != InvalidOid && expectedtypeid != SPI_gettypeid(rec->tupdesc, fno) ) {
!                      Datum cval = SPI_getbinval(rec->tup, rec->tupdesc, fno, isnull);
!                      cval =   exec_simple_cast_value(cval,
!                                      SPI_gettypeid(rec->tupdesc, fno),
!                                      expectedtypeid,
!                                      -1,
!                                      *isnull);
!
!                      *value = cval;
!                      *typeid = expectedtypeid;
!                      /* ereport(ERROR,
!                              (errcode(ERRCODE_DATATYPE_MISMATCH),
!                               errmsg("type of \"%s\" does not match that when preparing the plan",
!                                      rec->refname)));
!                      */
!                  }
!                  else { /* expected typeid matches */
!                      *value = SPI_getbinval(rec->tup, rec->tupdesc, fno, isnull);
!                      *typeid = SPI_gettypeid(rec->tupdesc, fno);
!                  }
!                  break;
!              }
!
!          case PLPGSQL_DTYPE_RECFIELDNAMES:
!              /* Construct array datum from record field names */
!              {
!                  Oid            arraytypeid,
!                              arrayelemtypeid = TEXTOID;
!                  int16            arraytyplen,
!                              elemtyplen;
!                  bool            elemtypbyval;
!                  char            elemtypalign;
!                  ArrayType        *arrayval;
!                  PLpgSQL_recfieldproperties * recfp = (PLpgSQL_recfieldproperties *) datum;
!                  PLpgSQL_rec        *rec = (PLpgSQL_rec *) (estate->datums[recfp->recparentno]);
!                  int            fc, tfc = 0;
!                  Datum            *arrayelems;
!                  char            *fieldname;
!
!                  if (!HeapTupleIsValid(rec->tup))
!                      ereport(ERROR,
!                        (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
!                         errmsg("record \"%s\" is not assigned yet",
!                                rec->refname),
!                         errdetail("The tuple structure of a not-yet-assigned record is indeterminate.")));
!                  arrayelems = palloc(sizeof(Datum) * rec->tupdesc->natts);
!                  arraytypeid = get_array_type(arrayelemtypeid);
!                  arraytyplen = get_typlen(arraytypeid);
!                  get_typlenbyvalalign(arrayelemtypeid,
!                               &elemtyplen,
!                               &elemtypbyval,
!                               &elemtypalign);
!
!                  if ( expectedtypeid != InvalidOid && expectedtypeid != arraytypeid )
!                      ereport(ERROR,
!                              (errcode(ERRCODE_DATATYPE_MISMATCH),
!                               errmsg("type of \"%s\" does not match array type when preparing the plan",
!                                      rec->refname)));
!                  for ( fc = 0; fc < rec->tupdesc->natts; ++fc ) {
!                      fieldname = SPI_fname(rec->tupdesc, fc+1);
!                      if ( fieldname ) {
!                          arrayelems[fc] = DirectFunctionCall1(textin, CStringGetDatum(fieldname));
!                          pfree(fieldname);
!                          ++tfc;
!                      }
!                  }
!                  arrayval = construct_array(arrayelems, tfc,
!                               arrayelemtypeid,
!                               elemtyplen,
!                               elemtypbyval,
!                               elemtypalign);
!
!
!                  /* construct_array copies data; free temp elem array */
!                  for ( fc = 0; fc < tfc; ++fc )
!                      pfree(DatumGetPointer(arrayelems[fc]));
!                  pfree(arrayelems);
!                  *value = PointerGetDatum(arrayval);
!                  *typeid = arraytypeid;
!                  *isnull = false;
!                  /* need to save the pointer because otherwise it does not get freed */
!                  if ( recfp->save_fieldnames )
!                      pfree(recfp->save_fieldnames);
!                  recfp->save_fieldnames = arrayval;
!                  break;
!              }
!
          case PLPGSQL_DTYPE_TRIGARG:
              {
                  PLpgSQL_trigarg *trigarg = (PLpgSQL_trigarg *) datum;
***************
*** 3617,3623 ****
       */
      if (expr->plan == NULL)
          exec_prepare_plan(estate, expr);
!
      /*
       * If this is a simple expression, bypass SPI and use the executor
       * directly
--- 3754,3782 ----
       */
      if (expr->plan == NULL)
          exec_prepare_plan(estate, expr);
!     else {
!         /*
!          * check for any subexpressions with varying type in the expression
!          * currently (July 05), this is a record field of a record indexed by a variable
!          */
!         int            i;
!         PLpgSQL_datum        *d;
!         PLpgSQL_recfield    *rf;
!         for ( i = 0; i < expr->nparams; ++i ) {
!             d = estate->datums[expr->params[i]];
!             if ( d->dtype == PLPGSQL_DTYPE_RECFIELD ) {
!                 rf = (PLpgSQL_recfield *)d;
!                 if ( rf->fieldindex_flag == RECFIELD_USE_INDEX_VAR )
!                     break;
!             }
!         }
!         if ( i < expr->nparams ) { /* expr may change it's type */
!             /* now discard the plan and get new one */
!             SPI_freeplan(expr->plan);
!             expr->plan = NULL;
!             exec_prepare_plan(estate, expr);
!         }
!     }
      /*
       * If this is a simple expression, bypass SPI and use the executor
       * directly
*** ./src/pl/plpgsql/src/pl_funcs.c.orig    Thu Mar  9 22:29:37 2006
--- ./src/pl/plpgsql/src/pl_funcs.c    Mon May  8 22:49:50 2006
***************
*** 1044,1052 ****
                  printf("REC %s\n", ((PLpgSQL_rec *) d)->refname);
                  break;
              case PLPGSQL_DTYPE_RECFIELD:
!                 printf("RECFIELD %-16s of REC %d\n",
!                        ((PLpgSQL_recfield *) d)->fieldname,
!                        ((PLpgSQL_recfield *) d)->recparentno);
                  break;
              case PLPGSQL_DTYPE_ARRAYELEM:
                  printf("ARRAYELEM of VAR %d subscript ",
--- 1044,1056 ----
                  printf("REC %s\n", ((PLpgSQL_rec *) d)->refname);
                  break;
              case PLPGSQL_DTYPE_RECFIELD:
!                 if ( ((PLpgSQL_recfield *) d)->fieldindex_flag == RECFIELD_USE_FIELDNAME )
!                     printf("RECFIELD %-16s of REC %d\n",
!                            ((PLpgSQL_recfield *) d)->fieldindex.fieldname,
!                            ((PLpgSQL_recfield *) d)->recparentno);
!                 else
!                     printf("RECFIELD Variable of REC %d\n",
!                            ((PLpgSQL_recfield *) d)->recparentno);
                  break;
              case PLPGSQL_DTYPE_ARRAYELEM:
                  printf("ARRAYELEM of VAR %d subscript ",
*** ./src/pl/plpgsql/src/plpgsql.h.orig    Thu Mar  9 22:29:38 2006
--- ./src/pl/plpgsql/src/plpgsql.h    Mon May  8 22:49:50 2006
***************
*** 52,58 ****
      PLPGSQL_DTYPE_RECFIELD,
      PLPGSQL_DTYPE_ARRAYELEM,
      PLPGSQL_DTYPE_EXPR,
!     PLPGSQL_DTYPE_TRIGARG
  };

  /* ----------
--- 52,59 ----
      PLPGSQL_DTYPE_RECFIELD,
      PLPGSQL_DTYPE_ARRAYELEM,
      PLPGSQL_DTYPE_EXPR,
!     PLPGSQL_DTYPE_TRIGARG,
!     PLPGSQL_DTYPE_RECFIELDNAMES
  };

  /* ----------
***************
*** 251,260 ****
  {                                /* Field in record */
      int            dtype;
      int            rfno;
!     char       *fieldname;
      int            recparentno;    /* dno of parent record */
  } PLpgSQL_recfield;


  typedef struct
  {                                /* Element of array variable */
--- 252,276 ----
  {                                /* Field in record */
      int            dtype;
      int            rfno;
!     union {
!         char    *fieldname;
!         int    indexvar_no;        /* dno of variable holding index string */
!     } fieldindex;
!     enum {
!         RECFIELD_USE_FIELDNAME,
!         RECFIELD_USE_INDEX_VAR,
!     }    fieldindex_flag;
      int            recparentno;    /* dno of parent record */
  } PLpgSQL_recfield;

+ typedef struct
+ {                                /* Field in record */
+     int            dtype;
+     int            rfno;
+     int            recparentno;            /* dno of parent record */
+     ArrayType *        save_fieldnames;
+ } PLpgSQL_recfieldproperties;
+

  typedef struct
  {                                /* Element of array variable */
***************
*** 659,664 ****
--- 675,682 ----
  extern int    plpgsql_parse_tripwordtype(char *word);
  extern int    plpgsql_parse_wordrowtype(char *word);
  extern int    plpgsql_parse_dblwordrowtype(char *word);
+ extern int    plpgsql_parse_recfieldnames(char *word);
+ extern int    plpgsql_parse_recindex(char *word);
  extern PLpgSQL_type *plpgsql_parse_datatype(const char *string);
  extern PLpgSQL_type *plpgsql_build_datatype(Oid typeOid, int32 typmod);
  extern PLpgSQL_variable *plpgsql_build_variable(const char *refname, int lineno,
*** ./src/pl/plpgsql/src/scan.l.orig    Thu Mar  9 22:29:38 2006
--- ./src/pl/plpgsql/src/scan.l    Mon May  8 22:49:50 2006
***************
*** 222,227 ****
--- 222,233 ----
  {param}{space}*\.{space}*{identifier}{space}*%ROWTYPE    {
      plpgsql_error_lineno = plpgsql_scanner_lineno();
      return plpgsql_parse_dblwordrowtype(yytext); }
+ {identifier}{space}*\.\(\*\)        {
+     plpgsql_error_lineno = plpgsql_scanner_lineno();
+     return plpgsql_parse_recfieldnames(yytext); }
+ {identifier}{space}*\.\({identifier}\)        {
+     plpgsql_error_lineno = plpgsql_scanner_lineno();
+     return plpgsql_parse_recindex(yytext); }

  {digit}+        { return T_NUMBER;            }

*** ./src/test/regress/expected/plpgsql.out.orig    Thu Mar 23 05:22:36 2006
--- ./src/test/regress/expected/plpgsql.out    Mon May  8 22:49:50 2006
***************
*** 2725,2730 ****
--- 2725,2768 ----
  $$ language plpgsql;
  ERROR:  end label "outer_label" specified for unlabelled block
  CONTEXT:  compile of PL/pgSQL function "end_label4" near line 5
+ -- check introspective records
+ create table ritest (i INT4, t TEXT);
+ insert into ritest (i, t) VALUES (1, 'sometext');
+ create function test_record() returns void as $$
+ declare
+   cname text;
+   tval  text;
+   ival  int4;
+   tval2 text;
+   ival2 int4;
+   columns text[];
+   r     RECORD;
+ begin
+   SELECT INTO r * FROM ritest WHERE i = 1;
+   ival := r.i;
+   tval := r.t;
+   RAISE NOTICE 'ival=%, tval=%', ival, tval;
+   cname := 'i';
+   ival2 := r.(cname);
+   cname :='t';
+   tval2 := r.(cname);
+   RAISE NOTICE 'ival2=%, tval2=%', ival2, tval2;
+   columns := r.(*);
+   RAISE NOTICE 'fieldnames=%', columns;
+   RETURN;
+ end;
+ $$ language plpgsql;
+ select test_record();
+ NOTICE:  ival=1, tval=sometext
+ NOTICE:  ival2=1, tval2=sometext
+ NOTICE:  fieldnames={i,t}
+  test_record
+ -------------
+
+  (1 row)
+
+ drop table ritest;
+ drop function test_record();
  -- using list of scalars in fori and fore stmts
  create function for_vect() returns void as $proc$
  <<lbl>>declare a integer; b varchar; c varchar; r record;
*** ./src/test/regress/sql/plpgsql.sql.orig    Thu Mar 23 05:22:37 2006
--- ./src/test/regress/sql/plpgsql.sql    Mon May  8 22:49:50 2006
***************
*** 2281,2286 ****
--- 2281,2318 ----
  end;
  $$ language plpgsql;

+ -- check introspective records
+ create table ritest (i INT4, t TEXT);
+ insert into ritest (i, t) VALUES (1, 'sometext');
+ create function test_record() returns void as $$
+ declare
+   cname text;
+   tval  text;
+   ival  int4;
+   tval2 text;
+   ival2 int4;
+   columns text[];
+   r     RECORD;
+ begin
+   SELECT INTO r * FROM ritest WHERE i = 1;
+   ival := r.i;
+   tval := r.t;
+   RAISE NOTICE 'ival=%, tval=%', ival, tval;
+   cname := 'i';
+   ival2 := r.(cname);
+   cname :='t';
+   tval2 := r.(cname);
+   RAISE NOTICE 'ival2=%, tval2=%', ival2, tval2;
+   columns := r.(*);
+   RAISE NOTICE 'fieldnames=%', columns;
+   RETURN;
+ end;
+ $$ language plpgsql;
+ select test_record();
+ drop table ritest;
+ drop function test_record();
+
+
  -- using list of scalars in fori and fore stmts
  create function for_vect() returns void as $proc$
  <<lbl>>declare a integer; b varchar; c varchar; r record;