Re: tablefunc functions in postgresql - Mailing list pgsql-patches

Attached is a patch for contrib/tablefunc. It fixes two issues raised by
Lars Boegild Thomsen (full email below) and also corrects the regression
expected output for a recent backend message adjustment. Please apply.

Thanks,

Joe


Lars Boegild Thomsen wrote:
> Dear Mr. Conway,
>
> I've noticed that you're the author of the tablefunc functions in the
> postgresql distribution.  I've come across a problem that I think is a bug.
> I realize that you're of course under no obligation to support this
> software, so I am just writing turn your attention to the problem - if
> indeed it is a problem :)
>
> I've go the following table defined:
>
> create table domains (
>     id int not null primary key default nextval('domain_id_seq'),
>     name varchar(64) not null,
>     description text,
>     owner_id int
> ) without oids;
>
> And for testing it contains the following:
>
> ipsmart=> select * from domains;
>  id |        name         |               description                |
> owner_id
> ----+---------------------+------------------------------------------+------
> ----
>   0 | justit.ws           | The overall owner of the IPSmart product |
>   1 | sp1.justit.ws       | Domain 1                                 |
> 0
>   2 | sp2.justit.ws       | Domain 2                                 |
> 0
>   3 | test1.sp1.justit.ws | Sub Domain 1                             |
> 1
>   4 | test2.sp1.justit.ws | Sub Domain 2                             |
> 1
>
> The owner_id column referenfes the id column of the same table.
>
> Now - I need to traverse this table from a certain level (and in production
> it will become rather deep).
>
> I've got two problems - one just an annoying "feature" of the connectby
> function and the other I think is a bug.  Let's start with the bug :)
> Traversing from node '0' is ok:
>
> ipsmart=> select * from connectby('domains', 'id', 'owner_id', '0', 0) as
> t(id int, owner_id int, level int);
>  id | owner_id | level
> ----+----------+-------
>   0 |          |     0
>   1 |        0 |     1
>   3 |        1 |     2
>   4 |        1 |     2
>   2 |        0 |     1
> (5 rows)
>
> No problem there.  Traversing from node 1 also works fine (results are 1, 3
> and 4).  But traversing from node 2 - nothing is returned.
>
> ipsmart=> select * from connectby('domains', 'id', 'owner_id', '2', 0) as
> t(id int, owner_id int, level int);
>  id | owner_id | level
> ----+----------+-------
> (0 rows)
>
> This is kind of weird since in the first two examples the root node was
> clearly returned as well.  Is this a feature or a bug?
>
> The other problem is not really a bug but rather a slight irritation :)
> Originally I had the 'owner_id" defined as "not null'.  I let the root refer
> to itself as parent and added the references constraint after I inserted the
> first record.  This way I could guarantee that the parent was always
> defined.  Unfortunately your "connectby" function wouldn't accept this and
> it lead to an infinite recurse.   Wouldn't it be reasonable easy to let your
> function stop if the node was referring to itself?
>
> Anyway - as mentioned earlier - I don't really expect that you've got time
> to look at this (although I would of course appreciate it a lot if you had).
> Even if you don't have time - I can live with the minor problems unless
> PostgreSQL get a real "connect by prior" construct :)
>
> Regards,
>
>     Lars...
>
> --
> Lars Boegild Thomsen
> Technical Director
> JustIT Sdn. Bhd.
> Cell Phone (MY): +60 (16) 323 1999
> ICQ: 6478559
> Yahoo Chat: lars_boegild_thomsen@yahoo.com
> MSN Chat: lars_boegild_thomsen@hotmail.com
> http://www.justit.ws
>

Index: contrib/tablefunc/tablefunc.c
===================================================================
RCS file: /opt/src/cvs/pgsql-server/contrib/tablefunc/tablefunc.c,v
retrieving revision 1.24
diff -c -r1.24 tablefunc.c
*** contrib/tablefunc/tablefunc.c    13 Sep 2003 21:44:50 -0000    1.24
--- contrib/tablefunc/tablefunc.c    1 Oct 2003 22:47:11 -0000
***************
*** 1295,1329 ****
      int            ret;
      int            proc;
      int            serial_column;

      if (max_depth > 0 && level > max_depth)
          return tupstore;

      /* Build initial sql statement */
      if (!show_serial)
      {
!         appendStringInfo(sql, "SELECT %s, %s FROM %s WHERE %s = '%s' AND %s IS NOT NULL",
                           key_fld,
                           parent_key_fld,
                           relname,
                           parent_key_fld,
                           start_with,
!                          key_fld);
          serial_column = 0;
      }
      else
      {
!         appendStringInfo(sql, "SELECT %s, %s FROM %s WHERE %s = '%s' AND %s IS NOT NULL ORDER BY %s",
                           key_fld,
                           parent_key_fld,
                           relname,
                           parent_key_fld,
                           start_with,
!                          key_fld,
                           orderby_fld);
          serial_column = 1;
      }

      /* Retrieve the desired rows */
      ret = SPI_exec(sql->data, 0);
      proc = SPI_processed;
--- 1295,1394 ----
      int            ret;
      int            proc;
      int            serial_column;
+     StringInfo    branchstr = NULL;
+     StringInfo    chk_branchstr = NULL;
+     StringInfo    chk_current_key = NULL;
+     char      **values;
+     char       *current_key;
+     char       *current_key_parent;
+     char        current_level[INT32_STRLEN];
+     char        serial_str[INT32_STRLEN];
+     char       *current_branch;
+     HeapTuple    tuple;

      if (max_depth > 0 && level > max_depth)
          return tupstore;

+     /* start a new branch */
+     branchstr = makeStringInfo();
+
+     /* need these to check for recursion */
+     chk_branchstr = makeStringInfo();
+     chk_current_key = makeStringInfo();
+
      /* Build initial sql statement */
      if (!show_serial)
      {
!         appendStringInfo(sql, "SELECT %s, %s FROM %s WHERE %s = '%s' AND %s IS NOT NULL AND %s <> %s",
                           key_fld,
                           parent_key_fld,
                           relname,
                           parent_key_fld,
                           start_with,
!                          key_fld, key_fld, parent_key_fld);
          serial_column = 0;
      }
      else
      {
!         appendStringInfo(sql, "SELECT %s, %s FROM %s WHERE %s = '%s' AND %s IS NOT NULL AND %s <> %s ORDER BY %s",
                           key_fld,
                           parent_key_fld,
                           relname,
                           parent_key_fld,
                           start_with,
!                          key_fld, key_fld, parent_key_fld,
                           orderby_fld);
          serial_column = 1;
      }

+     if (show_branch)
+         values = (char **) palloc((CONNECTBY_NCOLS + serial_column) * sizeof(char *));
+     else
+         values = (char **) palloc((CONNECTBY_NCOLS_NOBRANCH + serial_column) * sizeof(char *));
+
+     /* First time through, do a little setup */
+     if (level == 0)
+     {
+         /* root value is the one we initially start with */
+         values[0] = start_with;
+
+         /* root value has no parent */
+         values[1] = NULL;
+
+         /* root level is 0 */
+         sprintf(current_level, "%d", level);
+         values[2] = current_level;
+
+         /* root branch is just starting root value */
+         if (show_branch)
+             values[3] = start_with;
+
+         /* root starts the serial with 1 */
+         if (show_serial)
+         {
+             sprintf(serial_str, "%d", (*serial)++);
+             if (show_branch)
+                 values[4] = serial_str;
+             else
+                 values[3] = serial_str;
+         }
+
+         /* construct the tuple */
+         tuple = BuildTupleFromCStrings(attinmeta, values);
+
+         /* switch to long lived context while storing the tuple */
+         oldcontext = MemoryContextSwitchTo(per_query_ctx);
+
+         /* now store it */
+         tuplestore_puttuple(tupstore, tuple);
+
+         /* now reset the context */
+         MemoryContextSwitchTo(oldcontext);
+
+         /* increment level */
+         level++;
+     }
+
      /* Retrieve the desired rows */
      ret = SPI_exec(sql->data, 0);
      proc = SPI_processed;
***************
*** 1331,1364 ****
      /* Check for qualifying tuples */
      if ((ret == SPI_OK_SELECT) && (proc > 0))
      {
-         HeapTuple    tuple;
          HeapTuple    spi_tuple;
          SPITupleTable *tuptable = SPI_tuptable;
          TupleDesc    spi_tupdesc = tuptable->tupdesc;
          int            i;
-         char       *current_key;
-         char       *current_key_parent;
-         char        current_level[INT32_STRLEN];
-         char        serial_str[INT32_STRLEN];
-         char       *current_branch;
-         char      **values;
-         StringInfo    branchstr = NULL;
-         StringInfo    chk_branchstr = NULL;
-         StringInfo    chk_current_key = NULL;
-
-         /* start a new branch */
-         branchstr = makeStringInfo();
-
-         /* need these to check for recursion */
-         chk_branchstr = makeStringInfo();
-         chk_current_key = makeStringInfo();

!         if (show_branch)
!             values = (char **) palloc((CONNECTBY_NCOLS + serial_column) * sizeof(char *));
!         else
!             values = (char **) palloc((CONNECTBY_NCOLS_NOBRANCH + serial_column) * sizeof(char *));
!
!         /* First time through, do a little setup */
          if (level == 0)
          {
              /*
--- 1396,1407 ----
      /* Check for qualifying tuples */
      if ((ret == SPI_OK_SELECT) && (proc > 0))
      {
          HeapTuple    spi_tuple;
          SPITupleTable *tuptable = SPI_tuptable;
          TupleDesc    spi_tupdesc = tuptable->tupdesc;
          int            i;

!         /* First time through, do a little more setup */
          if (level == 0)
          {
              /*
***************
*** 1373,1417 ****
                           errmsg("invalid return type"),
                       errdetail("Return and SQL tuple descriptions are " \
                                 "incompatible.")));
-
-             /* root value is the one we initially start with */
-             values[0] = start_with;
-
-             /* root value has no parent */
-             values[1] = NULL;
-
-             /* root level is 0 */
-             sprintf(current_level, "%d", level);
-             values[2] = current_level;
-
-             /* root branch is just starting root value */
-             if (show_branch)
-                 values[3] = start_with;
-
-             /* root starts the serial with 1 */
-             if (show_serial)
-             {
-                 sprintf(serial_str, "%d", (*serial)++);
-                 if (show_branch)
-                     values[4] = serial_str;
-                 else
-                     values[3] = serial_str;
-             }
-
-             /* construct the tuple */
-             tuple = BuildTupleFromCStrings(attinmeta, values);
-
-             /* switch to long lived context while storing the tuple */
-             oldcontext = MemoryContextSwitchTo(per_query_ctx);
-
-             /* now store it */
-             tuplestore_puttuple(tupstore, tuple);
-
-             /* now reset the context */
-             MemoryContextSwitchTo(oldcontext);
-
-             /* increment level */
-             level++;
          }

          for (i = 0; i < proc; i++)
--- 1416,1421 ----
Index: contrib/tablefunc/expected/tablefunc.out
===================================================================
RCS file: /opt/src/cvs/pgsql-server/contrib/tablefunc/expected/tablefunc.out,v
retrieving revision 1.9
diff -c -r1.9 tablefunc.out
*** contrib/tablefunc/expected/tablefunc.out    13 Sep 2003 21:44:50 -0000    1.9
--- contrib/tablefunc/expected/tablefunc.out    1 Oct 2003 22:34:32 -0000
***************
*** 127,133 ****
  -- hash based crosstab
  --
  create table cth(id serial, rowid text, rowdt timestamp, attribute text, val text);
! NOTICE:  CREATE TABLE will create implicit sequence "cth_id_seq" for SERIAL column "cth.id"
  insert into cth values(DEFAULT,'test1','01 March 2003','temperature','42');
  insert into cth values(DEFAULT,'test1','01 March 2003','test_result','PASS');
  -- the next line is intentionally left commented and is therefore a "missing" attribute
--- 127,133 ----
  -- hash based crosstab
  --
  create table cth(id serial, rowid text, rowdt timestamp, attribute text, val text);
! NOTICE:  CREATE TABLE will create implicit sequence "cth_id_seq" for "serial" column "cth.id"
  insert into cth values(DEFAULT,'test1','01 March 2003','temperature','42');
  insert into cth values(DEFAULT,'test1','01 March 2003','test_result','PASS');
  -- the next line is intentionally left commented and is therefore a "missing" attribute