pg_upgrade: Transfer pg_largeobject_metadata's files when possible.

Commit 161a3e8b68 taught pg_upgrade to use COPY for large object
metadata for upgrades from v12 and newer, which is much faster to
restore than the proper large object commands.  For upgrades from
v16 and newer, we can take this a step further and transfer the
large object metadata files as if they were user tables.  We can't
transfer the files from older versions because the aclitem data
type (needed by pg_largeobject_metadata.lomacl) changed its storage
format in v16 (see commit 7b378237aa).  Note that this commit is
essentially a revert of commit 12a53c732c.

There are a couple of caveats.  First, we still need to COPY the
corresponding pg_shdepend rows for large objects.  Second, we need
to COPY anything in pg_largeobject_metadata with a comment or
security label, else restoring those will fail.  This means that an
upgrade in which every large object has a comment or security label
won't gain anything from this commit, but it should at least avoid
making those unusual use-cases any worse.

pg_upgrade must also take care to transfer the relfilenodes of
pg_largeobject_metadata and its index, as was done for
pg_largeobject in commits d498e052b4 and bbe08b8869.

Reviewed-by: Michael Paquier <michael@paquier.xyz>
Discussion: https://postgr.es/m/aJ3_Gih_XW1_O2HF%40nathan

src/backend/commands/tablecmds.c

@@ -42,6 +42,7 @@
 #include "catalog/pg_foreign_table.h"
 #include "catalog/pg_inherits.h"
 #include "catalog/pg_largeobject.h"
+#include "catalog/pg_largeobject_metadata.h"
 #include "catalog/pg_namespace.h"
 #include "catalog/pg_opclass.h"
 #include "catalog/pg_policy.h"
@@ -2389,12 +2390,15 @@ truncate_check_rel(Oid relid, Form_pg_class reltuple)
 	/*
 	 * Most system catalogs can't be truncated at all, or at least not unless
 	 * allow_system_table_mods=on. As an exception, however, we allow
-	 * pg_largeobject to be truncated as part of pg_upgrade, because we need
-	 * to change its relfilenode to match the old cluster, and allowing a
-	 * TRUNCATE command to be executed is the easiest way of doing that.
+	 * pg_largeobject and pg_largeobject_metadata to be truncated as part of
+	 * pg_upgrade, because we need to change its relfilenode to match the old
+	 * cluster, and allowing a TRUNCATE command to be executed is the easiest
+	 * way of doing that.
 	 */
 	if (!allowSystemTableMods && IsSystemClass(relid, reltuple)
-		&& (!IsBinaryUpgrade || relid != LargeObjectRelationId))
+		&& (!IsBinaryUpgrade ||
+			(relid != LargeObjectRelationId &&
+			 relid != LargeObjectMetadataRelationId)))
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 				 errmsg("permission denied: \"%s\" is a system catalog",

src/bin/pg_dump/pg_dump.c

@@ -1131,6 +1131,23 @@ main(int argc, char **argv)
 		shdepend->dataObj->filtercond = "WHERE classid = 'pg_largeobject'::regclass "
 			"AND dbid = (SELECT oid FROM pg_database "
 			"            WHERE datname = current_database())";
+
+		/*
+		 * If upgrading from v16 or newer, only dump large objects with
+		 * comments/seclabels.  For these upgrades, pg_upgrade can copy/link
+		 * pg_largeobject_metadata's files (which is usually faster) but we
+		 * still need to dump LOs with comments/seclabels here so that the
+		 * subsequent COMMENT and SECURITY LABEL commands work.  pg_upgrade
+		 * can't copy/link the files from older versions because aclitem
+		 * (needed by pg_largeobject_metadata.lomacl) changed its storage
+		 * format in v16.
+		 */
+		if (fout->remoteVersion >= 160000)
+			lo_metadata->dataObj->filtercond = "WHERE oid IN "
+				"(SELECT objoid FROM pg_description "
+				"WHERE classoid = " CppAsString2(LargeObjectRelationId) " "
+				"UNION SELECT objoid FROM pg_seclabel "
+				"WHERE classoid = " CppAsString2(LargeObjectRelationId) ")";
 	}
 
 	/*
@@ -3629,26 +3646,32 @@ dumpDatabase(Archive *fout)
 	/*
 	 * pg_largeobject comes from the old system intact, so set its
 	 * relfrozenxids, relminmxids and relfilenode.
+	 *
+	 * pg_largeobject_metadata also comes from the old system intact for
+	 * upgrades from v16 and newer, so set its relfrozenxids, relminmxids, and
+	 * relfilenode, too.  pg_upgrade can't copy/link the files from older
+	 * versions because aclitem (needed by pg_largeobject_metadata.lomacl)
+	 * changed its storage format in v16.
 	 */
 	if (dopt->binary_upgrade)
 	{
 		PGresult   *lo_res;
 		PQExpBuffer loFrozenQry = createPQExpBuffer();
 		PQExpBuffer loOutQry = createPQExpBuffer();
+		PQExpBuffer lomOutQry = createPQExpBuffer();
 		PQExpBuffer loHorizonQry = createPQExpBuffer();
+		PQExpBuffer lomHorizonQry = createPQExpBuffer();
 		int			ii_relfrozenxid,
 					ii_relfilenode,
 					ii_oid,
 					ii_relminmxid;
 
-		/*
-		 * pg_largeobject
-		 */
 		if (fout->remoteVersion >= 90300)
 			appendPQExpBuffer(loFrozenQry, "SELECT relfrozenxid, relminmxid, relfilenode, oid\n"
 							  "FROM pg_catalog.pg_class\n"
-							  "WHERE oid IN (%u, %u);\n",
-							  LargeObjectRelationId, LargeObjectLOidPNIndexId);
+							  "WHERE oid IN (%u, %u, %u, %u);\n",
+							  LargeObjectRelationId, LargeObjectLOidPNIndexId,
+							  LargeObjectMetadataRelationId, LargeObjectMetadataOidIndexId);
 		else
 			appendPQExpBuffer(loFrozenQry, "SELECT relfrozenxid, 0 AS relminmxid, relfilenode, oid\n"
 							  "FROM pg_catalog.pg_class\n"
@@ -3663,35 +3686,57 @@ dumpDatabase(Archive *fout)
 		ii_oid = PQfnumber(lo_res, "oid");
 
 		appendPQExpBufferStr(loHorizonQry, "\n-- For binary upgrade, set pg_largeobject relfrozenxid and relminmxid\n");
+		appendPQExpBufferStr(lomHorizonQry, "\n-- For binary upgrade, set pg_largeobject_metadata relfrozenxid and relminmxid\n");
 		appendPQExpBufferStr(loOutQry, "\n-- For binary upgrade, preserve pg_largeobject and index relfilenodes\n");
+		appendPQExpBufferStr(lomOutQry, "\n-- For binary upgrade, preserve pg_largeobject_metadata and index relfilenodes\n");
 		for (int i = 0; i < PQntuples(lo_res); ++i)
 		{
 			Oid			oid;
 			RelFileNumber relfilenumber;
+			PQExpBuffer horizonQry;
+			PQExpBuffer outQry;
+
+			oid = atooid(PQgetvalue(lo_res, i, ii_oid));
+			relfilenumber = atooid(PQgetvalue(lo_res, i, ii_relfilenode));
 
-			appendPQExpBuffer(loHorizonQry, "UPDATE pg_catalog.pg_class\n"
+			if (oid == LargeObjectRelationId ||
+				oid == LargeObjectLOidPNIndexId)
+			{
+				horizonQry = loHorizonQry;
+				outQry = loOutQry;
+			}
+			else
+			{
+				horizonQry = lomHorizonQry;
+				outQry = lomOutQry;
+			}
+
+			appendPQExpBuffer(horizonQry, "UPDATE pg_catalog.pg_class\n"
 							  "SET relfrozenxid = '%u', relminmxid = '%u'\n"
 							  "WHERE oid = %u;\n",
 							  atooid(PQgetvalue(lo_res, i, ii_relfrozenxid)),
 							  atooid(PQgetvalue(lo_res, i, ii_relminmxid)),
 							  atooid(PQgetvalue(lo_res, i, ii_oid)));
 
-			oid = atooid(PQgetvalue(lo_res, i, ii_oid));
-			relfilenumber = atooid(PQgetvalue(lo_res, i, ii_relfilenode));
-
-			if (oid == LargeObjectRelationId)
-				appendPQExpBuffer(loOutQry,
+			if (oid == LargeObjectRelationId ||
+				oid == LargeObjectMetadataRelationId)
+				appendPQExpBuffer(outQry,
 								  "SELECT pg_catalog.binary_upgrade_set_next_heap_relfilenode('%u'::pg_catalog.oid);\n",
 								  relfilenumber);
-			else if (oid == LargeObjectLOidPNIndexId)
-				appendPQExpBuffer(loOutQry,
+			else if (oid == LargeObjectLOidPNIndexId ||
+					 oid == LargeObjectMetadataOidIndexId)
+				appendPQExpBuffer(outQry,
 								  "SELECT pg_catalog.binary_upgrade_set_next_index_relfilenode('%u'::pg_catalog.oid);\n",
 								  relfilenumber);
 		}
 
 		appendPQExpBufferStr(loOutQry,
 							 "TRUNCATE pg_catalog.pg_largeobject;\n");
+		appendPQExpBufferStr(lomOutQry,
+							 "TRUNCATE pg_catalog.pg_largeobject_metadata;\n");
+
 		appendPQExpBufferStr(loOutQry, loHorizonQry->data);
+		appendPQExpBufferStr(lomOutQry, lomHorizonQry->data);
 
 		ArchiveEntry(fout, nilCatalogId, createDumpId(),
 					 ARCHIVE_OPTS(.tag = "pg_largeobject",
@@ -3699,11 +3744,20 @@ dumpDatabase(Archive *fout)
 								  .section = SECTION_PRE_DATA,
 								  .createStmt = loOutQry->data));
 
+		if (fout->remoteVersion >= 160000)
+			ArchiveEntry(fout, nilCatalogId, createDumpId(),
+						 ARCHIVE_OPTS(.tag = "pg_largeobject_metadata",
+									  .description = "pg_largeobject_metadata",
+									  .section = SECTION_PRE_DATA,
+									  .createStmt = lomOutQry->data));
+
 		PQclear(lo_res);
 
 		destroyPQExpBuffer(loFrozenQry);
 		destroyPQExpBuffer(loHorizonQry);
+		destroyPQExpBuffer(lomHorizonQry);
 		destroyPQExpBuffer(loOutQry);
+		destroyPQExpBuffer(lomOutQry);
 	}
 
 	PQclear(res);

src/bin/pg_upgrade/Makefile

@@ -3,8 +3,7 @@
 PGFILEDESC = "pg_upgrade - an in-place binary upgrade utility"
 PGAPPICON = win32
 
-# required for 003_upgrade_logical_replication_slots.pl
-EXTRA_INSTALL=contrib/test_decoding
+EXTRA_INSTALL=contrib/test_decoding src/test/modules/dummy_seclabel
 
 subdir = src/bin/pg_upgrade
 top_builddir = ../../..

src/bin/pg_upgrade/info.c

@@ -498,7 +498,10 @@ get_rel_infos_query(void)
 	 *
 	 * pg_largeobject contains user data that does not appear in pg_dump
 	 * output, so we have to copy that system table.  It's easiest to do that
-	 * by treating it as a user table.
+	 * by treating it as a user table.  We can do the same for
+	 * pg_largeobject_metadata for upgrades from v16 and newer.  pg_upgrade
+	 * can't copy/link the files from older versions because aclitem (needed
+	 * by pg_largeobject_metadata.lomacl) changed its storage format in v16.
 	 */
 	appendPQExpBuffer(&query,
 					  "WITH regular_heap (reloid, indtable, toastheap) AS ( "
@@ -514,10 +517,12 @@ get_rel_infos_query(void)
 					  "                        'binary_upgrade', 'pg_toast') AND "
 					  "      c.oid >= %u::pg_catalog.oid) OR "
 					  "     (n.nspname = 'pg_catalog' AND "
-					  "      relname IN ('pg_largeobject') ))), ",
+					  "      relname IN ('pg_largeobject'%s) ))), ",
 					  (user_opts.transfer_mode == TRANSFER_MODE_SWAP) ?
 					  ", " CppAsString2(RELKIND_SEQUENCE) : "",
-					  FirstNormalObjectId);
+					  FirstNormalObjectId,
+					  (GET_MAJOR_VERSION(old_cluster.major_version) >= 1600) ?
+					  ", 'pg_largeobject_metadata'" : "");
 
 	/*
 	 * Add a CTE that collects OIDs of toast tables belonging to the tables

src/bin/pg_upgrade/pg_upgrade.c

@@ -29,9 +29,9 @@
  *	We control all assignments of pg_enum.oid because these oids are stored
  *	in user tables as enum values.
  *
- *	We control all assignments of pg_authid.oid for historical reasons (the
- *	oids used to be stored in pg_largeobject_metadata, which is now copied via
- *	SQL commands), that might change at some point in the future.
+ *	We control all assignments of pg_authid.oid because the oids are stored in
+ *	pg_largeobject_metadata, which is copied via file transfer for upgrades
+ *	from v16 and newer.
  *
  *	We control all assignments of pg_database.oid because we want the directory
  *	names to match between the old and new cluster.

src/bin/pg_upgrade/t/006_transfer_modes.pl

@@ -45,6 +45,22 @@ sub test_mode
 		$old->append_conf('postgresql.conf', "allow_in_place_tablespaces = true");
 	}
 
+	# We can only test security labels if both the old and new installations
+	# have dummy_seclabel.
+	my $test_seclabel = 1;
+	$old->start;
+	if (!$old->check_extension('dummy_seclabel'))
+	{
+		$test_seclabel = 0;
+	}
+	$old->stop;
+	$new->start;
+	if (!$new->check_extension('dummy_seclabel'))
+	{
+		$test_seclabel = 0;
+	}
+	$new->stop;
+
 	# Create a small variety of simple test objects on the old cluster.  We'll
 	# check that these reach the new version after upgrading.
 	$old->start;
@@ -83,6 +99,29 @@ sub test_mode
 		$old->safe_psql('testdb3',
 			"CREATE TABLE test6 AS SELECT generate_series(607, 711)");
 	}
+
+	# While we are here, test handling of large objects.
+	$old->safe_psql('postgres', q|
+		CREATE ROLE regress_lo_1;
+		CREATE ROLE regress_lo_2;
+
+		SELECT lo_from_bytea(4532, '\xffffff00');
+		COMMENT ON LARGE OBJECT 4532 IS 'test';
+
+		SELECT lo_from_bytea(4533, '\x0f0f0f0f');
+		ALTER LARGE OBJECT 4533 OWNER TO regress_lo_1;
+		GRANT SELECT ON LARGE OBJECT 4533 TO regress_lo_2;
+	|);
+
+	if ($test_seclabel)
+	{
+		$old->safe_psql('postgres', q|
+			CREATE EXTENSION dummy_seclabel;
+
+			SELECT lo_from_bytea(4534, '\x00ffffff');
+			SECURITY LABEL ON LARGE OBJECT 4534 IS 'classified';
+		|);
+	}
 	$old->stop;
 
 	my $result = command_ok_or_fails_like(
@@ -132,6 +171,34 @@ sub test_mode
 			$result = $new->safe_psql('testdb3', "SELECT COUNT(*) FROM test6");
 			is($result, '105', "test6 data after pg_upgrade $mode");
 		}
+
+		# Tests for large objects
+		$result = $new->safe_psql('postgres', "SELECT lo_get(4532)");
+		is($result, '\xffffff00', "LO contents after upgrade");
+		$result = $new->safe_psql('postgres',
+			"SELECT obj_description(4532, 'pg_largeobject')");
+		is($result, 'test', "comment on LO after pg_upgrade");
+
+		$result = $new->safe_psql('postgres', "SELECT lo_get(4533)");
+		is($result, '\x0f0f0f0f', "LO contents after upgrade");
+		$result = $new->safe_psql('postgres',
+			"SELECT lomowner::regrole FROM pg_largeobject_metadata WHERE oid = 4533");
+		is($result, 'regress_lo_1', "LO owner after upgrade");
+		$result = $new->safe_psql('postgres',
+			"SELECT lomacl FROM pg_largeobject_metadata WHERE oid = 4533");
+		is($result, '{regress_lo_1=rw/regress_lo_1,regress_lo_2=r/regress_lo_1}',
+			"LO ACL after upgrade");
+
+		if ($test_seclabel)
+		{
+			$result = $new->safe_psql('postgres', "SELECT lo_get(4534)");
+			is($result, '\x00ffffff', "LO contents after upgrade");
+			$result = $new->safe_psql('postgres', q|
+				SELECT label FROM pg_seclabel WHERE objoid = 4534
+				AND classoid = 'pg_largeobject'::regclass
+			|);
+			is($result, 'classified', "seclabel on LO after pg_upgrade");
+		}
 		$new->stop;
 	}