In REASSIGN OWNED of a database, lock the tuple as mandated.

nmisch · nmisch · commit fa6131377054 · 2024-12-28T07:16:26.000-08:00
Commit aac2c9b mandated such locking and attempted to fulfill that mandate, but it missed REASSIGN OWNED. Hence, it remained possible to lose VACUUM's inplace update of datfrozenxid if a REASSIGN OWNED processed that database at the same time. This didn't affect the other inplace-updated catalog, pg_class. For pg_class, REASSIGN OWNED calls ATExecChangeOwner() instead of the generic AlterObjectOwner_internal(), and ATExecChangeOwner() fulfills the locking mandate. Like in GRANT, implement this by following the locking protocol for any catalog subject to the generic AlterObjectOwner_internal(). It would suffice to do this for IsInplaceUpdateOid() catalogs only. Back-patch to v13 (all supported versions). Kirill Reshke. Reported by Alexander Kukushkin. Discussion: https://postgr.es/m/CAFh8B=mpKjAy4Cuun-HP-f_vRzh2HSvYFG3rhVfYbfEBUhBAGg@mail.gmail.com
diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c
@@ -2779,14 +2779,35 @@ get_object_property_data(Oid class_id)
  */
 HeapTuple
 get_catalog_object_by_oid(Relation catalog, AttrNumber oidcol, Oid objectId)
+{
+	return
+		get_catalog_object_by_oid_extended(catalog, oidcol, objectId, false);
+}
+
+/*
+ * Same as get_catalog_object_by_oid(), but with an additional "locktup"
+ * argument controlling whether to acquire a LOCKTAG_TUPLE at mode
+ * InplaceUpdateTupleLock.  See README.tuplock section "Locking to write
+ * inplace-updated tables".
+ */
+HeapTuple
+get_catalog_object_by_oid_extended(Relation catalog,
+								   AttrNumber oidcol,
+								   Oid objectId,
+								   bool locktup)
 {
 	HeapTuple	tuple;
 	Oid			classId = RelationGetRelid(catalog);
 	int			oidCacheId = get_object_catcache_oid(classId);
 
 	if (oidCacheId > 0)
 	{
-		tuple = SearchSysCacheCopy1(oidCacheId, ObjectIdGetDatum(objectId));
+		if (locktup)
+			tuple = SearchSysCacheLockedCopy1(oidCacheId,
+											  ObjectIdGetDatum(objectId));
+		else
+			tuple = SearchSysCacheCopy1(oidCacheId,
+										ObjectIdGetDatum(objectId));
 		if (!HeapTupleIsValid(tuple))	/* should not happen */
 			return NULL;
 	}
@@ -2811,6 +2832,10 @@ get_catalog_object_by_oid(Relation catalog, AttrNumber oidcol, Oid objectId)
 			systable_endscan(scan);
 			return NULL;
 		}
+
+		if (locktup)
+			LockTuple(catalog, &tuple->t_self, InplaceUpdateTupleLock);
+
 		tuple = heap_copytuple(tuple);
 
 		systable_endscan(scan);
diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
@@ -59,6 +59,7 @@
 #include "miscadmin.h"
 #include "replication/logicalworker.h"
 #include "rewrite/rewriteDefine.h"
+#include "storage/lmgr.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
@@ -931,7 +932,9 @@ AlterObjectOwner_internal(Oid classId, Oid objectId, Oid new_ownerId)
 
 	rel = table_open(catalogId, RowExclusiveLock);
 
-	oldtup = get_catalog_object_by_oid(rel, Anum_oid, objectId);
+	/* Search tuple and lock it. */
+	oldtup =
+		get_catalog_object_by_oid_extended(rel, Anum_oid, objectId, true);
 	if (oldtup == NULL)
 		elog(ERROR, "cache lookup failed for object %u of catalog \"%s\"",
 			 objectId, RelationGetRelationName(rel));
@@ -1031,6 +1034,8 @@ AlterObjectOwner_internal(Oid classId, Oid objectId, Oid new_ownerId)
 		/* Perform actual update */
 		CatalogTupleUpdate(rel, &newtup->t_self, newtup);
 
+		UnlockTuple(rel, &oldtup->t_self, InplaceUpdateTupleLock);
+
 		/* Update owner dependency reference */
 		changeDependencyOnOwner(classId, objectId, new_ownerId);
 
@@ -1039,6 +1044,8 @@ AlterObjectOwner_internal(Oid classId, Oid objectId, Oid new_ownerId)
 		pfree(nulls);
 		pfree(replaces);
 	}
+	else
+		UnlockTuple(rel, &oldtup->t_self, InplaceUpdateTupleLock);
 
 	/* Note the post-alter hook gets classId not catalogId */
 	InvokeObjectPostAlterHook(classId, objectId, 0);
diff --git a/src/include/catalog/objectaddress.h b/src/include/catalog/objectaddress.h
@@ -69,6 +69,10 @@ extern bool get_object_namensp_unique(Oid class_id);
 
 extern HeapTuple get_catalog_object_by_oid(Relation catalog,
 										   AttrNumber oidcol, Oid objectId);
+extern HeapTuple get_catalog_object_by_oid_extended(Relation catalog,
+													AttrNumber oidcol,
+													Oid objectId,
+													bool locktup);
 
 extern char *getObjectDescription(const ObjectAddress *object,
 								  bool missing_ok);
diff --git a/src/test/regress/expected/database.out b/src/test/regress/expected/database.out
@@ -12,4 +12,10 @@ WHERE datname = 'regression_utf8';
 -- load catcache entry, if nothing else does
 ALTER DATABASE regression_utf8 RESET TABLESPACE;
 ROLLBACK;
+CREATE ROLE regress_datdba_before;
+CREATE ROLE regress_datdba_after;
+ALTER DATABASE regression_utf8 OWNER TO regress_datdba_before;
+REASSIGN OWNED BY regress_datdba_before TO regress_datdba_after;
 DROP DATABASE regression_utf8;
+DROP ROLE regress_datdba_before;
+DROP ROLE regress_datdba_after;
diff --git a/src/test/regress/sql/database.sql b/src/test/regress/sql/database.sql
@@ -14,4 +14,11 @@ WHERE datname = 'regression_utf8';
 ALTER DATABASE regression_utf8 RESET TABLESPACE;
 ROLLBACK;
 
+CREATE ROLE regress_datdba_before;
+CREATE ROLE regress_datdba_after;
+ALTER DATABASE regression_utf8 OWNER TO regress_datdba_before;
+REASSIGN OWNED BY regress_datdba_before TO regress_datdba_after;
+
 DROP DATABASE regression_utf8;
+DROP ROLE regress_datdba_before;
+DROP ROLE regress_datdba_after;
