]> git.ipfire.org Git - thirdparty/postgresql.git/commitdiff
Handle arrays and ranges in pg_upgrade's test for non-upgradable types.
authorTom Lane <tgl@sss.pgh.pa.us>
Wed, 13 Nov 2019 16:35:37 +0000 (11:35 -0500)
committerTom Lane <tgl@sss.pgh.pa.us>
Wed, 13 Nov 2019 16:35:37 +0000 (11:35 -0500)
pg_upgrade needs to check whether certain non-upgradable data types
appear anywhere on-disk in the source cluster.  It knew that it has
to check for these types being contained inside domains and composite
types; but it somehow overlooked that they could be contained in
arrays and ranges, too.  Extend the existing recursive-containment
query to handle those cases.

We probably should have noticed this oversight while working on
commit 0ccfc2822 and follow-ups, but we failed to :-(.  The whole
thing's possibly a bit overdesigned, since we don't really expect
that any of these types will appear on disk; but if we're going to
the effort of doing a recursive search then it's silly not to cover
all the possibilities.

While at it, refactor so that we have only one copy of the search
logic, not three-and-counting.  Also, to keep the branches looking
more alike, back-patch the output wording change of commit 1634d3615.

Back-patch to all supported branches.

Discussion: https://postgr.es/m/31473.1573412838@sss.pgh.pa.us

contrib/pg_upgrade/version.c

index 9b0301d3d4a81bf799c5b73e070fad30ae9460bf..d611bb1fa86f6086afb1f10afae3523452ebd11e 100644 (file)
@@ -98,27 +98,27 @@ new_9_0_populate_pg_largeobject_metadata(ClusterInfo *cluster, bool check_mode)
 
 
 /*
- * old_9_3_check_for_line_data_type_usage()
- *     9.3 -> 9.4
- *     Fully implement the 'line' data type in 9.4, which previously returned
- *     "not enabled" by default and was only functionally enabled with a
- *     compile-time switch;  9.4 "line" has different binary and text
- *     representation formats;  checks tables and indexes.
+ * check_for_data_type_usage
+ *     Detect whether there are any stored columns depending on the given type
+ *
+ * If so, write a report to the given file name, and return true.
+ *
+ * We check for the type in tables, matviews, and indexes, but not views;
+ * there's no storage involved in a view.
  */
-void
-old_9_3_check_for_line_data_type_usage(ClusterInfo *cluster)
+static bool
+check_for_data_type_usage(ClusterInfo *cluster, const char *typename,
+                                                 char *output_path)
 {
-       int                     dbnum;
-       FILE       *script = NULL;
        bool            found = false;
-       char            output_path[MAXPGPATH];
-
-       prep_status("Checking for invalid \"line\" user columns");
-
-       snprintf(output_path, sizeof(output_path), "tables_using_line.txt");
+       FILE       *script = NULL;
+       int                     dbnum;
 
        for (dbnum = 0; dbnum < cluster->dbarr.ndbs; dbnum++)
        {
+               DbInfo     *active_db = &cluster->dbarr.dbs[dbnum];
+               PGconn     *conn = connectToServer(cluster, active_db->db_name);
+               PQExpBufferData querybuf;
                PGresult   *res;
                bool            db_used = false;
                int                     ntups;
@@ -126,50 +126,68 @@ old_9_3_check_for_line_data_type_usage(ClusterInfo *cluster)
                int                     i_nspname,
                                        i_relname,
                                        i_attname;
-               DbInfo     *active_db = &cluster->dbarr.dbs[dbnum];
-               PGconn     *conn = connectToServer(cluster, active_db->db_name);
 
                /*
-                * The pg_catalog.line type may be wrapped in a domain or composite
-                * type, or both (9.3 did not allow domains on composite types, but
-                * there may be multi-level composite type). To detect these cases
-                * we need a recursive CTE.
+                * The type of interest might be wrapped in a domain, array,
+                * composite, or range, and these container types can be nested (to
+                * varying extents depending on server version, but that's not of
+                * concern here).  To handle all these cases we need a recursive CTE.
                 */
-               res = executeQueryOrDie(conn,
-                                                               "WITH RECURSIVE oids AS ( "
-               /* the pg_catalog.line type itself */
-                                                               "       SELECT 'pg_catalog.line'::pg_catalog.regtype AS oid "
-                                                               "       UNION ALL "
-                                                               "       SELECT * FROM ( "
-               /* domains on the type */
-                                                               "               WITH x AS (SELECT oid FROM oids) "
-                                                               "                       SELECT t.oid FROM pg_catalog.pg_type t, x WHERE typbasetype = x.oid AND typtype = 'd' "
-                                                               "                       UNION "
-               /* composite types containing the type */
-                                                               "                       SELECT t.oid FROM pg_catalog.pg_type t, pg_catalog.pg_class c, pg_catalog.pg_attribute a, x "
-                                                               "                       WHERE t.typtype = 'c' AND "
-                                                               "                                 t.oid = c.reltype AND "
-                                                               "                                 c.oid = a.attrelid AND "
-                                                               "                                 NOT a.attisdropped AND "
-                                                               "                                 a.atttypid = x.oid "
-                                                               "       ) foo "
-                                                               ") "
-                                                               "SELECT n.nspname, c.relname, a.attname "
-                                                               "FROM   pg_catalog.pg_class c, "
-                                                               "               pg_catalog.pg_namespace n, "
-                                                               "               pg_catalog.pg_attribute a "
-                                                               "WHERE  c.oid = a.attrelid AND "
-                                                               "               NOT a.attisdropped AND "
-                                                               "               a.atttypid IN (SELECT oid FROM oids) AND "
-                                                               "               c.relkind IN ("
-                                                               CppAsString2(RELKIND_RELATION) ", "
-                                                               CppAsString2(RELKIND_MATVIEW) ", "
-                                                               CppAsString2(RELKIND_INDEX) ") AND "
-                                                               "               c.relnamespace = n.oid AND "
+               initPQExpBuffer(&querybuf);
+               appendPQExpBuffer(&querybuf,
+                                                 "WITH RECURSIVE oids AS ( "
+               /* the target type itself */
+                                                 "     SELECT '%s'::pg_catalog.regtype AS oid "
+                                                 "     UNION ALL "
+                                                 "     SELECT * FROM ( "
+               /* inner WITH because we can only reference the CTE once */
+                                                 "             WITH x AS (SELECT oid FROM oids) "
+               /* domains on any type selected so far */
+                                                 "                     SELECT t.oid FROM pg_catalog.pg_type t, x WHERE typbasetype = x.oid AND typtype = 'd' "
+                                                 "                     UNION ALL "
+               /* arrays over any type selected so far */
+                                                 "                     SELECT t.oid FROM pg_catalog.pg_type t, x WHERE typelem = x.oid AND typtype = 'b' "
+                                                 "                     UNION ALL "
+               /* composite types containing any type selected so far */
+                                                 "                     SELECT t.oid FROM pg_catalog.pg_type t, pg_catalog.pg_class c, pg_catalog.pg_attribute a, x "
+                                                 "                     WHERE t.typtype = 'c' AND "
+                                                 "                               t.oid = c.reltype AND "
+                                                 "                               c.oid = a.attrelid AND "
+                                                 "                               NOT a.attisdropped AND "
+                                                 "                               a.atttypid = x.oid ",
+                                                 typename);
+
+               /* Ranges came in in 9.2 */
+               if (GET_MAJOR_VERSION(cluster->major_version) >= 902)
+                       appendPQExpBuffer(&querybuf,
+                                                         "                     UNION ALL "
+                       /* ranges containing any type selected so far */
+                                                         "                     SELECT t.oid FROM pg_catalog.pg_type t, pg_catalog.pg_range r, x "
+                                                         "                     WHERE t.typtype = 'r' AND r.rngtypid = t.oid AND r.rngsubtype = x.oid");
+
+               appendPQExpBuffer(&querybuf,
+                                                 "     ) foo "
+                                                 ") "
+               /* now look for stored columns of any such type */
+                                                 "SELECT n.nspname, c.relname, a.attname "
+                                                 "FROM pg_catalog.pg_class c, "
+                                                 "             pg_catalog.pg_namespace n, "
+                                                 "             pg_catalog.pg_attribute a "
+                                                 "WHERE        c.oid = a.attrelid AND "
+                                                 "             NOT a.attisdropped AND "
+                                                 "             a.atttypid IN (SELECT oid FROM oids) AND "
+                                                 "             c.relkind IN ("
+                                                 CppAsString2(RELKIND_RELATION) ", "
+                                                 CppAsString2(RELKIND_MATVIEW) ", "
+                                                 CppAsString2(RELKIND_INDEX) ") AND "
+                                                 "             c.relnamespace = n.oid AND "
                /* exclude possible orphaned temp tables */
-                                                               "               n.nspname !~ '^pg_temp_' AND "
-                                                "              n.nspname !~ '^pg_toast_temp_' AND "
-                                                               "               n.nspname NOT IN ('pg_catalog', 'information_schema')");
+                                                 "             n.nspname !~ '^pg_temp_' AND "
+                                                 "             n.nspname !~ '^pg_toast_temp_' AND "
+               /* exclude system catalogs, too */
+                                                 "             n.nspname NOT IN ('pg_catalog', 'information_schema')");
+
+               res = executeQueryOrDie(conn, "%s", querybuf.data);
 
                ntups = PQntuples(res);
                i_nspname = PQfnumber(res, "nspname");
@@ -182,7 +200,7 @@ old_9_3_check_for_line_data_type_usage(ClusterInfo *cluster)
                                pg_fatal("could not open file \"%s\": %s\n", output_path, getErrorText());
                        if (!db_used)
                        {
-                               fprintf(script, "Database: %s\n", active_db->db_name);
+                               fprintf(script, "In database: %s\n", active_db->db_name);
                                db_used = true;
                        }
                        fprintf(script, "  %s.%s.%s\n",
@@ -193,13 +211,36 @@ old_9_3_check_for_line_data_type_usage(ClusterInfo *cluster)
 
                PQclear(res);
 
+               termPQExpBuffer(&querybuf);
+
                PQfinish(conn);
        }
 
        if (script)
                fclose(script);
 
-       if (found)
+       return found;
+}
+
+
+/*
+ * old_9_3_check_for_line_data_type_usage()
+ *     9.3 -> 9.4
+ *     Fully implement the 'line' data type in 9.4, which previously returned
+ *     "not enabled" by default and was only functionally enabled with a
+ *     compile-time switch; as of 9.4 "line" has a different on-disk
+ *     representation format.
+ */
+void
+old_9_3_check_for_line_data_type_usage(ClusterInfo *cluster)
+{
+       char            output_path[MAXPGPATH];
+
+       prep_status("Checking for invalid \"line\" user columns");
+
+       snprintf(output_path, sizeof(output_path), "tables_using_line.txt");
+
+       if (check_for_data_type_usage(cluster, "pg_catalog.line", output_path))
        {
                pg_log(PG_REPORT, "fatal\n");
                pg_fatal("Your installation contains the \"line\" data type in user tables.  This\n"