updated WIP: arrays of composites - Mailing list pgsql-patches

Attached is my rework of David Fetter's array of composites patch. It
has all the agreed modifications and checks, except altering the name
mangling.

If people prefer I can apply this as it stands and then get working on
the name mangling piece. Or I can hold off.

I'm still wondering if we can get away without a catalog change on that
- e.g. could we look up an array type by looking for a pg_type entry
containing the base type's oid in the typelem field? Or would that be
too slow?

cheers

andrew
Index: doc/src/sgml/array.sgml
===================================================================
RCS file: /cvsroot/pgsql/doc/src/sgml/array.sgml,v
retrieving revision 1.60
diff -c -r1.60 array.sgml
*** doc/src/sgml/array.sgml    6 Apr 2007 19:22:38 -0000    1.60
--- doc/src/sgml/array.sgml    7 May 2007 17:17:29 -0000
***************
*** 11,17 ****
    <productname>PostgreSQL</productname> allows columns of a table to be
    defined as variable-length multidimensional arrays. Arrays of any
    built-in or user-defined base type or enum type can be created.
!   (Arrays of composite types or domains are not yet supported, however.)
   </para>

   <sect2>
--- 11,18 ----
    <productname>PostgreSQL</productname> allows columns of a table to be
    defined as variable-length multidimensional arrays. Arrays of any
    built-in or user-defined base type or enum type can be created.
!   Arrays of composite types are now supported.  Arrays of domains are
!   not yet supported.
   </para>

   <sect2>
Index: src/backend/catalog/heap.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/catalog/heap.c,v
retrieving revision 1.318
diff -c -r1.318 heap.c
*** src/backend/catalog/heap.c    2 Apr 2007 03:49:37 -0000    1.318
--- src/backend/catalog/heap.c    7 May 2007 17:17:31 -0000
***************
*** 45,50 ****
--- 45,51 ----
  #include "catalog/pg_statistic.h"
  #include "catalog/pg_type.h"
  #include "commands/tablecmds.h"
+ #include "commands/typecmds.h"
  #include "miscadmin.h"
  #include "optimizer/clauses.h"
  #include "optimizer/var.h"
***************
*** 402,417 ****
      char        att_typtype = get_typtype(atttypid);

      /*
!      * Warn user, but don't fail, if column to be created has UNKNOWN type
!      * (usually as a result of a 'retrieve into' - jolly)
       *
!      * Refuse any attempt to create a pseudo-type column.
       */
!     if (atttypid == UNKNOWNOID)
          ereport(WARNING,
                  (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
                   errmsg("column \"%s\" has type \"unknown\"", attname),
                   errdetail("Proceeding with relation creation anyway.")));
      else if (att_typtype == TYPTYPE_PSEUDO)
      {
          /* Special hack for pg_statistic: allow ANYARRAY during initdb */
--- 403,442 ----
      char        att_typtype = get_typtype(atttypid);

      /*
!      * For a composite type, recurse into its attributes. Otherwise,
       *
!      * a) warn user, but don't fail, if column to be created has UNKNOWN type
!      *    (usually as a result of a 'retrieve into' - jolly)
!      *
!      * b) Refuse any attempt to create a pseudo-type column, except for
!      *    pg_statistic hack.
       */
!     if (att_typtype == TYPTYPE_COMPOSITE)
!     {
!         Relation relation;
!         TupleDesc tupdesc;
!         int i;
!
!         relation = RelationIdGetRelation(get_typ_typrelid(atttypid));
!
!         tupdesc = RelationGetDescr(relation);
!
!         for (i = 0; i < tupdesc->natts; i++)
!         {
!             if (tupdesc->attrs[i]->attisdropped)
!                 continue;
!             CheckAttributeType(attname, tupdesc->attrs[i]->atttypid);
!         }
!
!         RelationClose(relation);
!     }
!     else if (atttypid == UNKNOWNOID)
!     {
          ereport(WARNING,
                  (errcode(ERRCODE_INVALID_TABLE_DEFINITION),
                   errmsg("column \"%s\" has type \"unknown\"", attname),
                   errdetail("Proceeding with relation creation anyway.")));
+     }
      else if (att_typtype == TYPTYPE_PSEUDO)
      {
          /* Special hack for pg_statistic: allow ANYARRAY during initdb */
***************
*** 763,768 ****
--- 788,794 ----
      Relation    pg_class_desc;
      Relation    new_rel_desc;
      Oid            new_type_oid;
+     char       *relarrayname;

      pg_class_desc = heap_open(RelationRelationId, RowExclusiveLock);

***************
*** 815,820 ****
--- 841,880 ----
                                        relnamespace,
                                        relid,
                                        relkind);
+     /*
+      * Add in the corresponding array types if appropriate.
+      */
+     if (IsUnderPostmaster && (relkind == RELKIND_RELATION ||
+                               relkind == RELKIND_VIEW ||
+                               relkind == RELKIND_COMPOSITE_TYPE))
+     {
+         relarrayname = makeArrayTypeName(relname);
+         TypeCreate(relarrayname,        /* Array type name */
+                    relnamespace,        /* Same namespace as parent */
+                    InvalidOid,            /* Not composite, so no relationOid  */
+                    0,                    /* relkind, also N/A here */
+                    -1,                    /* Internal size, unlimited */
+                    TYPTYPE_BASE,        /* Not a complex type - typelem is */
+                    DEFAULT_TYPDELIM,    /* Use the default */
+                    F_ARRAY_IN,            /* Macro for array input procedure */
+                    F_ARRAY_OUT,            /* Macro for array output procedure */
+                    F_ARRAY_RECV,        /* Macro for array receive (binary input) procedure */
+                    F_ARRAY_SEND,        /* Macro for array send (binary output) procedure */
+                    InvalidOid,            /* No input typmod */
+                    InvalidOid,            /* No output typmod */
+                    InvalidOid,            /* Default ANALYZE procedure */
+                    new_type_oid,        /* The OID just created */
+                    InvalidOid,            /* No base type--this isn't a DOMAIN */
+                    NULL,                /* No default type value */
+                    NULL,                /* Don't send binary */
+                    false,                /* Never passed by value */
+                    'd',                    /* Type align. Largest for safety */
+                    'x',                    /* Always TOASTable */
+                    -1,                    /* No typMod for non-domain types. */
+                    0,                    /* Array diminsions of typbasetype */
+                    false);                /* Type NOT NULL */
+         pfree(relarrayname);    /* Seems like the right thing to do here. */
+     }

      /*
       * now create an entry in pg_class for the relation.
Index: src/backend/commands/tablecmds.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/commands/tablecmds.c,v
retrieving revision 1.219
diff -c -r1.219 tablecmds.c
*** src/backend/commands/tablecmds.c    8 Apr 2007 01:26:32 -0000    1.219
--- src/backend/commands/tablecmds.c    7 May 2007 17:17:36 -0000
***************
*** 287,298 ****
      Datum        reloptions;
      ListCell   *listptr;
      AttrNumber    attnum;

      /*
!      * Truncate relname to appropriate length (probably a waste of time, as
!      * parser should have done this already).
       */
!     StrNCpy(relname, stmt->relation->relname, NAMEDATALEN);

      /*
       * Check consistency of arguments
--- 287,309 ----
      Datum        reloptions;
      ListCell   *listptr;
      AttrNumber    attnum;
+     char       *relarrayname;

      /*
!      * Truncate relname to appropriate length (probably a waste of time, as *
!      * parser should have done this already).  Because tables and views now get
!      * an array type, this depends on the relkind.
       */
!     if (relkind == RELKIND_RELATION ||
!         relkind == RELKIND_VIEW ||
!         relkind == RELKIND_COMPOSITE_TYPE)
!     {
!         StrNCpy(relname, stmt->relation->relname, NAMEDATALEN-2);
!     }
!     else
!     {
!         StrNCpy(relname, stmt->relation->relname, NAMEDATALEN);
!     }

      /*
       * Check consistency of arguments
***************
*** 6461,6468 ****

      AlterRelationNamespaceInternal(classRel, relid, oldNspOid, nspOid, true);

!     /* Fix the table's rowtype too */
      AlterTypeNamespaceInternal(rel->rd_rel->reltype, nspOid, false);

      /* Fix other dependent stuff */
      if (rel->rd_rel->relkind == RELKIND_RELATION)
--- 6472,6480 ----

      AlterRelationNamespaceInternal(classRel, relid, oldNspOid, nspOid, true);

!     /* Fix the table's types too */
      AlterTypeNamespaceInternal(rel->rd_rel->reltype, nspOid, false);
+     AlterTypeNamespaceInternal(get_array_type(rel->rd_rel->reltype), nspOid, false);

      /* Fix other dependent stuff */
      if (rel->rd_rel->relkind == RELKIND_RELATION)
Index: src/backend/commands/typecmds.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/commands/typecmds.c,v
retrieving revision 1.101
diff -c -r1.101 typecmds.c
*** src/backend/commands/typecmds.c    2 Apr 2007 03:49:38 -0000    1.101
--- src/backend/commands/typecmds.c    7 May 2007 17:17:37 -0000
***************
*** 474,479 ****
--- 474,480 ----
      Oid            typeoid;
      HeapTuple    tup;
      ObjectAddress object;
+     Form_pg_type typ;

      /* Make a TypeName so we can use standard type lookup machinery */
      typename = makeTypeNameFromNameList(names);
***************
*** 505,513 ****
      if (!HeapTupleIsValid(tup))
          elog(ERROR, "cache lookup failed for type %u", typeoid);

      /* Permission check: must own type or its namespace */
      if (!pg_type_ownercheck(typeoid, GetUserId()) &&
!       !pg_namespace_ownercheck(((Form_pg_type) GETSTRUCT(tup))->typnamespace,
                                 GetUserId()))
          aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_TYPE,
                         TypeNameToString(typename));
--- 506,524 ----
      if (!HeapTupleIsValid(tup))
          elog(ERROR, "cache lookup failed for type %u", typeoid);

+     typ = (Form_pg_type) GETSTRUCT(tup);
+
+     /* don't alow direct deletion of array types */
+     if (typ->typelem != 0 && typ->typlen == -1)
+             ereport(ERROR,
+                     (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
+                      errmsg("cannot delete array type \"%s\" ",
+                             TypeNameToString(typename))));
+
+
      /* Permission check: must own type or its namespace */
      if (!pg_type_ownercheck(typeoid, GetUserId()) &&
!       !pg_namespace_ownercheck(typ->typnamespace,
                                 GetUserId()))
          aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_TYPE,
                         TypeNameToString(typename));
***************
*** 2249,2254 ****
--- 2260,2273 ----
          elog(ERROR, "cache lookup failed for type %u", typeOid);
      typTup = (Form_pg_type) GETSTRUCT(tup);

+     /* don't allow direct alteration of array types */
+     if (typTup->typelem != 0 && typTup->typlen == -1)
+             ereport(ERROR,
+                     (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
+                      errmsg("cannot alter array type \"%s\" ",
+                             TypeNameToString(typename))));
+
+
      /*
       * If it's a composite type, we need to check that it really is a
       * free-standing composite type, and not a table's underlying type. We
***************
*** 2352,2357 ****
--- 2371,2379 ----
      TypeName   *typename;
      Oid            typeOid;
      Oid            nspOid;
+     Relation    rel;
+     HeapTuple    tup;
+     Form_pg_type typ;

      /* Make a TypeName so we can use standard type lookup machinery */
      typename = makeTypeNameFromNameList(names);
***************
*** 2365,2372 ****
--- 2387,2420 ----
      /* get schema OID and check its permissions */
      nspOid = LookupCreationNamespace(newschema);

+     /* don't allow direct alteration of array types */
+
+     rel = heap_open(TypeRelationId, RowExclusiveLock);
+
+     tup = SearchSysCacheCopy(TYPEOID,
+                              ObjectIdGetDatum(typeOid),
+                              0, 0, 0);
+     if (!HeapTupleIsValid(tup))
+         elog(ERROR, "cache lookup failed for type %u", typeOid);
+
+     typ = (Form_pg_type) GETSTRUCT(tup);
+
+     if (typ->typelem != 0 && typ->typlen == -1)
+             ereport(ERROR,
+                     (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
+                      errmsg("cannot alter array type \"%s\" ",
+                             TypeNameToString(typename))));
+
+
+     heap_freetuple(tup);
+
+     heap_close(rel, RowExclusiveLock);
+
      /* and do the work */
      AlterTypeNamespaceInternal(typeOid, nspOid, true);
+     typeOid = get_array_type(typeOid);
+     if (typeOid != InvalidOid)
+         AlterTypeNamespaceInternal(typeOid, nspOid, true);
  }

  /*
***************
*** 2431,2442 ****

      /* Detect whether type is a composite type (but not a table rowtype) */
      isCompositeType =
!         (typform->typtype == TYPTYPE_COMPOSITE &&
           get_rel_relkind(typform->typrelid) == RELKIND_COMPOSITE_TYPE);

      /* Enforce not-table-type if requested */
      if (typform->typtype == TYPTYPE_COMPOSITE && !isCompositeType &&
!         errorOnTableType)
          ereport(ERROR,
                  (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                   errmsg("%s is a table's row type",
--- 2479,2490 ----

      /* Detect whether type is a composite type (but not a table rowtype) */
      isCompositeType =
!         (typform->typtype == TYPTYPE_COMPOSITE && typform->typrelid != InvalidOid &&
           get_rel_relkind(typform->typrelid) == RELKIND_COMPOSITE_TYPE);

      /* Enforce not-table-type if requested */
      if (typform->typtype == TYPTYPE_COMPOSITE && !isCompositeType &&
!         typform->typrelid != InvalidOid && errorOnTableType)
          ereport(ERROR,
                  (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                   errmsg("%s is a table's row type",
***************
*** 2457,2463 ****
       * We need to modify the pg_class tuple as well to reflect the change of
       * schema.
       */
!     if (isCompositeType)
      {
          Relation    classRel;

--- 2505,2511 ----
       * We need to modify the pg_class tuple as well to reflect the change of
       * schema.
       */
!     if (isCompositeType && typform->typrelid != InvalidOid)
      {
          Relation    classRel;

***************
*** 2490,2496 ****
           * Update dependency on schema, if any --- a table rowtype has not got
           * one.
           */
!         if (typform->typtype != TYPTYPE_COMPOSITE)
              if (changeDependencyFor(TypeRelationId, typeOid,
                                  NamespaceRelationId, oldNspOid, nspOid) != 1)
                  elog(ERROR, "failed to change schema dependency for type %s",
--- 2538,2544 ----
           * Update dependency on schema, if any --- a table rowtype has not got
           * one.
           */
!         if (typform->typtype != TYPTYPE_COMPOSITE || typform->typrelid == InvalidOid)
              if (changeDependencyFor(TypeRelationId, typeOid,
                                  NamespaceRelationId, oldNspOid, nspOid) != 1)
                  elog(ERROR, "failed to change schema dependency for type %s",
Index: src/test/regress/expected/alter_table.out
===================================================================
RCS file: /cvsroot/pgsql/src/test/regress/expected/alter_table.out,v
retrieving revision 1.101
diff -c -r1.101 alter_table.out
*** src/test/regress/expected/alter_table.out    14 Feb 2007 01:58:58 -0000    1.101
--- src/test/regress/expected/alter_table.out    7 May 2007 17:17:41 -0000
***************
*** 1456,1468 ****
--- 1456,1471 ----

  -- clean up
  drop schema alter2 cascade;
+ NOTICE:  drop cascades to type alter2.ctype[]
  NOTICE:  drop cascades to composite type alter2.ctype
  NOTICE:  drop cascades to type alter2.ctype
  NOTICE:  drop cascades to type alter2.posint
  NOTICE:  drop cascades to function alter2.plus1(integer)
+ NOTICE:  drop cascades to type alter2.v1[]
  NOTICE:  drop cascades to view alter2.v1
  NOTICE:  drop cascades to rule _RETURN on view alter2.v1
  NOTICE:  drop cascades to sequence alter2.t1_f1_seq
  NOTICE:  drop cascades to default for table alter2.t1 column f1
+ NOTICE:  drop cascades to type alter2.t1[]
  NOTICE:  drop cascades to table alter2.t1
  NOTICE:  drop cascades to constraint t1_f2_check on table alter2.t1
Index: src/test/regress/expected/type_sanity.out
===================================================================
RCS file: /cvsroot/pgsql/src/test/regress/expected/type_sanity.out,v
retrieving revision 1.29
diff -c -r1.29 type_sanity.out
*** src/test/regress/expected/type_sanity.out    2 Apr 2007 03:49:42 -0000    1.29
--- src/test/regress/expected/type_sanity.out    7 May 2007 17:17:41 -0000
***************
*** 49,55 ****
  -- or basic types that do.
  SELECT p1.oid, p1.typname
  FROM pg_type as p1
! WHERE (p1.typtype = 'c' AND p1.typrelid = 0) OR
      (p1.typtype != 'c' AND p1.typrelid != 0);
   oid | typname
  -----+---------
--- 49,55 ----
  -- or basic types that do.
  SELECT p1.oid, p1.typname
  FROM pg_type as p1
! WHERE (p1.typtype = 'c' AND p1.typrelid = 0 AND p1.typname !~ '^_') OR
      (p1.typtype != 'c' AND p1.typrelid != 0);
   oid | typname
  -----+---------
Index: src/test/regress/sql/type_sanity.sql
===================================================================
RCS file: /cvsroot/pgsql/src/test/regress/sql/type_sanity.sql,v
retrieving revision 1.29
diff -c -r1.29 type_sanity.sql
*** src/test/regress/sql/type_sanity.sql    2 Apr 2007 03:49:42 -0000    1.29
--- src/test/regress/sql/type_sanity.sql    7 May 2007 17:17:41 -0000
***************
*** 46,52 ****

  SELECT p1.oid, p1.typname
  FROM pg_type as p1
! WHERE (p1.typtype = 'c' AND p1.typrelid = 0) OR
      (p1.typtype != 'c' AND p1.typrelid != 0);

  -- Look for basic or enum types that don't have an array type.
--- 46,52 ----

  SELECT p1.oid, p1.typname
  FROM pg_type as p1
! WHERE (p1.typtype = 'c' AND p1.typrelid = 0 AND p1.typname !~ '^_') OR
      (p1.typtype != 'c' AND p1.typrelid != 0);

  -- Look for basic or enum types that don't have an array type.