Re: Removing pg_migrator limitations - Mailing list pgsql-hackers

Bruce Momjian wrote:
> Tom Lane wrote:
> > Bruce Momjian <bruce@momjian.us> writes:
> > > ... The idea I had was to create a global structure:
> >
> > >     struct pg_migrator_oids {
> > >         Oid    pg_type;
> > >         Oid    pg_type_array;
> > >         ...
> > >     }
> >
> > > This would initialize to zero as a global structure, and only
> > > pg_migrator server-side functions set it.
> >
> > I would prefer *not* to do that, as that makes the list of settable oids
> > far more public than I would like; also you are totally dependent on
> > pg_migrator and the backend to be in sync about the definition of that
> > struct, which is going to be problematic in alpha releases in
> > particular, since PG_VERSION isn't going to distinguish them.
> >
> > What I had in mind was more like
> >
> >     static Oid next_pg_class_oid = InvalidOid;
> >
> >     void
> >     set_next_pg_class_oid(Oid oid)
> >     {
> >         next_pg_class_oid = oid;
> >     }
>
> Good point about requiring a link to a symbol;  a structure offset would
> not link to anything and would silently fail.
>
> Does exporting a function buy us anything vs. exporting a variable?
>
> > in each module that needs to be able to accept a next-oid setting,
> > and then the pg_migrator loadable module would expose SQL-callable
> > wrappers for these functions.  That way, any inconsistency shows up as
> > a link error: function needed not present.
>
> I will work on a patch to accomplish this, and have pg_migrator link in
> the .so only if the new server is >= 8.5, which allows a single
> pg_migrator binary to work for migration to 8.4 and 8.5.

I have completed the attached patch which assigns oids for all pg_type
rows when pg_dump --binary-upgrade is used.  This allows user-defined
arrays and composite types to be migrated cleanly.  I tested a reload of
the regression database with --binary-upgrade and all the pg_type oids
were identical.  The pg_migrator changes required to use this feature
are trivial.

The remaining issue is pg_enum oids.  Because it will be difficult to
pass an arbitrary number of oids into the backend, the idea was to
assign each enum value separately.  If we implement this TODO item:

    Allow adding/renaming/removing enumerated values to an existing
    enumerated data type

Particularly the "adding" part rather than the "renaming/removing" part,
pg_dump can create an enum type with one (or zero perhaps) enums, and
then use a pg_enum oid-setting function and then use ALTER TYPE ADD
ENUM to add each new value.

Comments?

--
  Bruce Momjian  <bruce@momjian.us>        http://momjian.us
  EnterpriseDB                             http://enterprisedb.com

  + If your life is a hard drive, Christ can be your backup. +
Index: src/backend/catalog/heap.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/catalog/heap.c,v
retrieving revision 1.361
diff -c -c -r1.361 heap.c
*** src/backend/catalog/heap.c    7 Dec 2009 05:22:21 -0000    1.361
--- src/backend/catalog/heap.c    23 Dec 2009 18:48:11 -0000
***************
*** 1001,1013 ****
      if (IsUnderPostmaster && (relkind == RELKIND_RELATION ||
                                relkind == RELKIND_VIEW ||
                                relkind == RELKIND_COMPOSITE_TYPE))
!     {
!         /* OK, so pre-assign a type OID for the array type */
!         Relation    pg_type = heap_open(TypeRelationId, AccessShareLock);
!
!         new_array_oid = GetNewOid(pg_type);
!         heap_close(pg_type, AccessShareLock);
!     }

      /*
       * Since defining a relation also defines a complex type, we add a new
--- 1001,1007 ----
      if (IsUnderPostmaster && (relkind == RELKIND_RELATION ||
                                relkind == RELKIND_VIEW ||
                                relkind == RELKIND_COMPOSITE_TYPE))
!         new_array_oid = AssignTypeArrayOid();

      /*
       * Since defining a relation also defines a complex type, we add a new
Index: src/backend/catalog/pg_type.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/catalog/pg_type.c,v
retrieving revision 1.127
diff -c -c -r1.127 pg_type.c
*** src/backend/catalog/pg_type.c    16 Aug 2009 18:14:34 -0000    1.127
--- src/backend/catalog/pg_type.c    23 Dec 2009 18:48:11 -0000
***************
*** 32,37 ****
--- 32,38 ----
  #include "utils/rel.h"
  #include "utils/syscache.h"

+ Oid binary_upgrade_next_pg_type_oid = InvalidOid;

  /* ----------------------------------------------------------------
   *        TypeShellMake
***************
*** 119,124 ****
--- 120,131 ----
       */
      tup = heap_form_tuple(tupDesc, values, nulls);

+     if (OidIsValid(binary_upgrade_next_pg_type_oid))
+     {
+         HeapTupleSetOid(tup, binary_upgrade_next_pg_type_oid);
+         binary_upgrade_next_pg_type_oid = InvalidOid;
+     }
+
      /*
       * insert the tuple in the relation and get the tuple's oid.
       */
***************
*** 409,418 ****
                                values,
                                nulls);

!         /* Force the OID if requested by caller, else heap_insert does it */
          if (OidIsValid(newTypeOid))
              HeapTupleSetOid(tup, newTypeOid);
!
          typeObjectId = simple_heap_insert(pg_type_desc, tup);
      }

--- 416,431 ----
                                values,
                                nulls);

!         /* Force the OID if requested by caller */
          if (OidIsValid(newTypeOid))
              HeapTupleSetOid(tup, newTypeOid);
!         else if (OidIsValid(binary_upgrade_next_pg_type_oid))
!         {
!             HeapTupleSetOid(tup, binary_upgrade_next_pg_type_oid);
!             binary_upgrade_next_pg_type_oid = InvalidOid;
!         }
!         /* else allow system to assign oid */
!
          typeObjectId = simple_heap_insert(pg_type_desc, tup);
      }

Index: src/backend/catalog/toasting.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/catalog/toasting.c,v
retrieving revision 1.22
diff -c -c -r1.22 toasting.c
*** src/backend/catalog/toasting.c    23 Dec 2009 02:35:18 -0000    1.22
--- src/backend/catalog/toasting.c    23 Dec 2009 18:48:11 -0000
***************
*** 31,36 ****
--- 31,37 ----
  #include "utils/builtins.h"
  #include "utils/syscache.h"

+ Oid binary_upgrade_next_pg_type_toast_oid = InvalidOid;

  static bool create_toast_table(Relation rel, Oid toastOid, Oid toastIndexOid,
                     Datum reloptions, bool force);
***************
*** 121,126 ****
--- 122,128 ----
      Relation    class_rel;
      Oid            toast_relid;
      Oid            toast_idxid;
+     Oid            toast_typid = InvalidOid;
      Oid            namespaceid;
      char        toast_relname[NAMEDATALEN];
      char        toast_idxname[NAMEDATALEN];
***************
*** 199,209 ****
      else
          namespaceid = PG_TOAST_NAMESPACE;

      toast_relid = heap_create_with_catalog(toast_relname,
                                             namespaceid,
                                             rel->rd_rel->reltablespace,
                                             toastOid,
!                                            InvalidOid,
                                             rel->rd_rel->relowner,
                                             tupdesc,
                                             NIL,
--- 201,217 ----
      else
          namespaceid = PG_TOAST_NAMESPACE;

+     if (OidIsValid(binary_upgrade_next_pg_type_toast_oid))
+     {
+         toast_typid = binary_upgrade_next_pg_type_toast_oid;
+         binary_upgrade_next_pg_type_toast_oid = InvalidOid;
+     }
+
      toast_relid = heap_create_with_catalog(toast_relname,
                                             namespaceid,
                                             rel->rd_rel->reltablespace,
                                             toastOid,
!                                            toast_typid,
                                             rel->rd_rel->relowner,
                                             tupdesc,
                                             NIL,
Index: src/backend/commands/typecmds.c
===================================================================
RCS file: /cvsroot/pgsql/src/backend/commands/typecmds.c,v
retrieving revision 1.140
diff -c -c -r1.140 typecmds.c
*** src/backend/commands/typecmds.c    19 Dec 2009 00:47:57 -0000    1.140
--- src/backend/commands/typecmds.c    23 Dec 2009 18:48:11 -0000
***************
*** 74,79 ****
--- 74,80 ----
      /* atts[] is of allocated length RelationGetNumberOfAttributes(rel) */
  } RelToCheck;

+ Oid binary_upgrade_next_pg_type_array_oid = InvalidOid;

  static Oid    findTypeInputFunction(List *procname, Oid typeOid);
  static Oid    findTypeOutputFunction(List *procname, Oid typeOid);
***************
*** 143,149 ****
      Oid            array_oid;
      Oid            typoid;
      Oid            resulttype;
-     Relation    pg_type;
      ListCell   *pl;

      /*
--- 144,149 ----
***************
*** 522,531 ****
                         NameListToString(analyzeName));
  #endif

!     /* Preassign array type OID so we can insert it in pg_type.typarray */
!     pg_type = heap_open(TypeRelationId, AccessShareLock);
!     array_oid = GetNewOid(pg_type);
!     heap_close(pg_type, AccessShareLock);

      /*
       * now have TypeCreate do all the real work.
--- 522,528 ----
                         NameListToString(analyzeName));
  #endif

!     array_oid = AssignTypeArrayOid();

      /*
       * now have TypeCreate do all the real work.
***************
*** 1101,1107 ****
      AclResult    aclresult;
      Oid            old_type_oid;
      Oid            enumArrayOid;
-     Relation    pg_type;

      /* Convert list of names to a name and namespace */
      enumNamespace = QualifiedNameGetCreationNamespace(stmt->typeName,
--- 1098,1103 ----
***************
*** 1129,1138 ****
                       errmsg("type \"%s\" already exists", enumName)));
      }

!     /* Preassign array type OID so we can insert it in pg_type.typarray */
!     pg_type = heap_open(TypeRelationId, AccessShareLock);
!     enumArrayOid = GetNewOid(pg_type);
!     heap_close(pg_type, AccessShareLock);

      /* Create the pg_type entry */
      enumTypeOid =
--- 1125,1131 ----
                       errmsg("type \"%s\" already exists", enumName)));
      }

!     enumArrayOid = AssignTypeArrayOid();

      /* Create the pg_type entry */
      enumTypeOid =
***************
*** 1470,1475 ****
--- 1463,1495 ----
      return procOid;
  }

+ /*
+  *    AssignTypeArrayOid
+  *
+  *    Pre-assign the type's array OID for use in pg_type.typarray
+  */
+ Oid
+ AssignTypeArrayOid(void)
+ {
+     Oid        type_array_oid;
+
+     /* Pre-assign the type's array OID for use in pg_type.typarray */
+     if (OidIsValid(binary_upgrade_next_pg_type_array_oid))
+     {
+         type_array_oid = binary_upgrade_next_pg_type_array_oid;
+         binary_upgrade_next_pg_type_array_oid = InvalidOid;
+     }
+     else
+     {
+         Relation    pg_type = heap_open(TypeRelationId, AccessShareLock);
+
+         type_array_oid = GetNewOid(pg_type);
+         heap_close(pg_type, AccessShareLock);
+     }
+
+     return type_array_oid;
+ }
+

  /*-------------------------------------------------------------------
   * DefineCompositeType
Index: src/bin/pg_dump/pg_dump.c
===================================================================
RCS file: /cvsroot/pgsql/src/bin/pg_dump/pg_dump.c,v
retrieving revision 1.560
diff -c -c -r1.560 pg_dump.c
*** src/bin/pg_dump/pg_dump.c    23 Dec 2009 04:10:50 -0000    1.560
--- src/bin/pg_dump/pg_dump.c    23 Dec 2009 18:48:12 -0000
***************
*** 196,201 ****
--- 196,206 ----
  static void dumpDatabase(Archive *AH);
  static void dumpEncoding(Archive *AH);
  static void dumpStdStrings(Archive *AH);
+ static void binary_upgrade_set_type_oids_by_type_oid(
+                     PQExpBuffer upgrade_buffer, Oid pg_type_oid);
+ static bool binary_upgrade_set_type_oids_by_rel_oid(
+                     PQExpBuffer upgrade_buffer, Oid pg_rel_oid);
+ static void binary_upgrade_clear_pg_type_toast_oid(PQExpBuffer upgrade_buffer);
  static const char *getAttrName(int attrnum, TableInfo *tblInfo);
  static const char *fmtCopyColumnList(const TableInfo *ti);
  static void do_sql_command(PGconn *conn, const char *query);
***************
*** 2176,2181 ****
--- 2181,2311 ----
      return 1;
  }

+ static void
+ binary_upgrade_set_type_oids_by_type_oid(PQExpBuffer upgrade_buffer,
+                                                Oid pg_type_oid)
+ {
+     PQExpBuffer upgrade_query = createPQExpBuffer();
+     int            ntups;
+     PGresult   *upgrade_res;
+     Oid            pg_type_array_oid;
+
+     appendPQExpBuffer(upgrade_buffer, "\n-- For binary upgrade, must preserve pg_type oid\n");
+     appendPQExpBuffer(upgrade_buffer,
+         "SELECT binary_upgrade.set_next_pg_type_oid('%u'::pg_catalog.oid);\n\n",
+         pg_type_oid);
+
+     /* we only support old >= 8.3 for binary upgrades */
+     appendPQExpBuffer(upgrade_query,
+                       "SELECT typarray "
+                       "FROM pg_catalog.pg_type "
+                       "WHERE pg_type.oid = '%u'::pg_catalog.oid;",
+                       pg_type_oid);
+
+     upgrade_res = PQexec(g_conn, upgrade_query->data);
+     check_sql_result(upgrade_res, g_conn, upgrade_query->data, PGRES_TUPLES_OK);
+
+     /* Expecting a single result only */
+     ntups = PQntuples(upgrade_res);
+     if (ntups != 1)
+     {
+         write_msg(NULL, ngettext("query returned %d row instead of one: %s\n",
+                                "query returned %d rows instead of one: %s\n",
+                                  ntups),
+                   ntups, upgrade_query->data);
+         exit_nicely();
+     }
+
+     pg_type_array_oid = atooid(PQgetvalue(upgrade_res, 0, PQfnumber(upgrade_res, "typarray")));
+
+     if (OidIsValid(pg_type_array_oid))
+     {
+         appendPQExpBuffer(upgrade_buffer,
+                             "\n-- For binary upgrade, must preserve pg_type array oid\n");
+         appendPQExpBuffer(upgrade_buffer,
+             "SELECT binary_upgrade.set_next_pg_type_array_oid('%u'::pg_catalog.oid);\n\n",
+             pg_type_array_oid);
+     }
+
+     PQclear(upgrade_res);
+     destroyPQExpBuffer(upgrade_query);
+ }
+
+ static bool
+ binary_upgrade_set_type_oids_by_rel_oid(PQExpBuffer upgrade_buffer,
+                                                Oid pg_rel_oid)
+ {
+     PQExpBuffer upgrade_query = createPQExpBuffer();
+     int            ntups;
+     PGresult   *upgrade_res;
+     Oid            pg_type_oid;
+     bool        toast_set = false;
+
+     /* we only support old >= 8.3 for binary upgrades */
+     appendPQExpBuffer(upgrade_query,
+                       "SELECT c.reltype AS crel, t.reltype AS trel "
+                       "FROM pg_catalog.pg_class c "
+                       "LEFT JOIN pg_catalog.pg_class t ON "
+                       "  (c.reltoastrelid = t.oid) "
+                       "WHERE c.oid = '%u'::pg_catalog.oid;",
+                       pg_rel_oid);
+
+     upgrade_res = PQexec(g_conn, upgrade_query->data);
+     check_sql_result(upgrade_res, g_conn, upgrade_query->data, PGRES_TUPLES_OK);
+
+     /* Expecting a single result only */
+     ntups = PQntuples(upgrade_res);
+     if (ntups != 1)
+     {
+         write_msg(NULL, ngettext("query returned %d row instead of one: %s\n",
+                                "query returned %d rows instead of one: %s\n",
+                                  ntups),
+                   ntups, upgrade_query->data);
+         exit_nicely();
+     }
+
+     pg_type_oid = atooid(PQgetvalue(upgrade_res, 0, PQfnumber(upgrade_res, "crel")));
+
+     binary_upgrade_set_type_oids_by_type_oid(upgrade_buffer, pg_type_oid);
+
+     if (!PQgetisnull(upgrade_res, 0, PQfnumber(upgrade_res, "trel")))
+     {
+         /* Toast tables do not have pg_type array rows */
+         Oid pg_type_toast_oid = atooid(PQgetvalue(upgrade_res, 0,
+                                         PQfnumber(upgrade_res, "trel")));
+
+         appendPQExpBuffer(upgrade_buffer, "\n-- For binary upgrade, must preserve pg_type toast oid\n");
+         appendPQExpBuffer(upgrade_buffer,
+             "SELECT binary_upgrade.set_next_pg_type_toast_oid('%u'::pg_catalog.oid);\n\n",
+             pg_type_toast_oid);
+
+         toast_set = true;
+     }
+
+     PQclear(upgrade_res);
+     destroyPQExpBuffer(upgrade_query);
+
+     return toast_set;
+ }
+
+ static void
+ binary_upgrade_clear_pg_type_toast_oid(PQExpBuffer upgrade_buffer)
+ {
+     /*
+      *    One complexity is that while the heap might now have a TOAST table,
+      *    the TOAST table might have been created long after creation when
+      *    the table was loaded with wide data.  For that reason, we clear
+      *    binary_upgrade_set_next_pg_type_toast_oid so it is not reused
+      *    by a later table.  Logically any later creation that needs a TOAST
+      *    table should have its own TOAST pg_type oid, but we are cautious.
+      */
+     appendPQExpBuffer(upgrade_buffer,
+         "\n-- For binary upgrade, clear toast oid because it might not have been needed\n");
+     appendPQExpBuffer(upgrade_buffer,
+         "SELECT binary_upgrade.set_next_pg_type_oid('%u'::pg_catalog.oid);\n\n",
+         InvalidOid);
+ }
+
  /*
   * getNamespaces:
   *      read all namespaces in the system catalogs and return them in the
***************
*** 6428,6433 ****
--- 6558,6567 ----
                        fmtId(tyinfo->dobj.namespace->dobj.name));
      appendPQExpBuffer(delq, "%s;\n",
                        fmtId(tyinfo->dobj.name));
+
+     if (binary_upgrade)
+         binary_upgrade_set_type_oids_by_type_oid(q, tyinfo->dobj.catId.oid);
+
      appendPQExpBuffer(q, "CREATE TYPE %s AS ENUM (\n",
                        fmtId(tyinfo->dobj.name));
      for (i = 0; i < num; i++)
***************
*** 6723,6728 ****
--- 6857,6866 ----
      appendPQExpBuffer(delq, "%s CASCADE;\n",
                        fmtId(tyinfo->dobj.name));

+     /* We might already have a shell type, but setting pg_type_oid is harmless */
+     if (binary_upgrade)
+         binary_upgrade_set_type_oids_by_type_oid(q, tyinfo->dobj.catId.oid);
+
      appendPQExpBuffer(q,
                        "CREATE TYPE %s (\n"
                        "    INTERNALLENGTH = %s",
***************
*** 6892,6897 ****
--- 7030,7038 ----
      else
          typdefault = NULL;

+     if (binary_upgrade)
+         binary_upgrade_set_type_oids_by_type_oid(q, tyinfo->dobj.catId.oid);
+
      appendPQExpBuffer(q,
                        "CREATE DOMAIN %s AS %s",
                        fmtId(tyinfo->dobj.name),
***************
*** 7002,7007 ****
--- 7143,7151 ----
      i_attname = PQfnumber(res, "attname");
      i_atttypdefn = PQfnumber(res, "atttypdefn");

+     if (binary_upgrade)
+         binary_upgrade_set_type_oids_by_type_oid(q, tyinfo->dobj.catId.oid);
+
      appendPQExpBuffer(q, "CREATE TYPE %s AS (",
                        fmtId(tyinfo->dobj.name));

***************
*** 7191,7196 ****
--- 7335,7344 ----
       * after it's filled in, otherwise the backend complains.
       */

+     if (binary_upgrade)
+         binary_upgrade_set_type_oids_by_type_oid(q,
+                                 stinfo->baseType->dobj.catId.oid);
+
      appendPQExpBuffer(q, "CREATE TYPE %s;\n",
                        fmtId(stinfo->dobj.name));

***************
*** 10226,10235 ****
      char       *storage;
      int            j,
                  k;
!
      /* Make sure we are in proper schema */
      selectSourceSchema(tbinfo->dobj.namespace->dobj.name);

      /* Is it a table or a view? */
      if (tbinfo->relkind == RELKIND_VIEW)
      {
--- 10374,10388 ----
      char       *storage;
      int            j,
                  k;
!     bool        toast_set = false;
!
      /* Make sure we are in proper schema */
      selectSourceSchema(tbinfo->dobj.namespace->dobj.name);

+     if (binary_upgrade)
+         toast_set = binary_upgrade_set_type_oids_by_rel_oid(q,
+                                                 tbinfo->dobj.catId.oid);
+
      /* Is it a table or a view? */
      if (tbinfo->relkind == RELKIND_VIEW)
      {
***************
*** 10606,10611 ****
--- 10759,10767 ----
          }
      }

+     if (binary_upgrade && toast_set)
+         binary_upgrade_clear_pg_type_toast_oid(q);
+
      ArchiveEntry(fout, tbinfo->dobj.catId, tbinfo->dobj.dumpId,
                   tbinfo->dobj.name,
                   tbinfo->dobj.namespace->dobj.name,
***************
*** 10617,10622 ****
--- 10773,10779 ----
                   tbinfo->dobj.dependencies, tbinfo->dobj.nDeps,
                   NULL, NULL);

+
      /* Dump Table Comments */
      dumpTableComment(fout, tbinfo, reltypename);

***************
*** 11235,11240 ****
--- 11392,11401 ----
                            fmtId(tbinfo->dobj.name));

          resetPQExpBuffer(query);
+
+         if (binary_upgrade)
+             binary_upgrade_set_type_oids_by_rel_oid(query, tbinfo->dobj.catId.oid);
+
          appendPQExpBuffer(query,
                            "CREATE SEQUENCE %s\n",
                            fmtId(tbinfo->dobj.name));
***************
*** 11270,11275 ****
--- 11431,11438 ----

          appendPQExpBuffer(query, ";\n");

+         /* binary_upgrade:  no need to clear TOAST table oid */
+
          ArchiveEntry(fout, tbinfo->dobj.catId, tbinfo->dobj.dumpId,
                       tbinfo->dobj.name,
                       tbinfo->dobj.namespace->dobj.name,
Index: src/include/commands/typecmds.h
===================================================================
RCS file: /cvsroot/pgsql/src/include/commands/typecmds.h,v
retrieving revision 1.25
diff -c -c -r1.25 typecmds.h
*** src/include/commands/typecmds.h    1 Jan 2009 17:23:58 -0000    1.25
--- src/include/commands/typecmds.h    23 Dec 2009 18:48:12 -0000
***************
*** 25,30 ****
--- 25,31 ----
  extern void DefineDomain(CreateDomainStmt *stmt);
  extern void DefineEnum(CreateEnumStmt *stmt);
  extern Oid    DefineCompositeType(const RangeVar *typevar, List *coldeflist);
+ extern Oid    AssignTypeArrayOid(void);

  extern void AlterDomainDefault(List *names, Node *defaultRaw);
  extern void AlterDomainNotNull(List *names, bool notNull);