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

Commit aac2c9b4fde889d13f859c233c2523345e72d32b 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
index 3a4e12592ff4fd0cf0e055ee9de15cf294648c21..060a1d826329faaf2ade8deab48e2c89c9c952eb 100644 (file)
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -2808,6 +2808,22 @@ 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);
@@ -2815,7 +2831,12 @@ get_catalog_object_by_oid(Relation catalog, AttrNumber oidcol, Oid objectId)
 
        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;
        }
@@ -2840,6 +2861,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
index aaae9f482fa46aa7dcea9b63851e688481929012..8cd1f5b102baf255bc35707ee58ab69ec5b59d7a 100644 (file)
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -60,6 +60,7 @@
 #include "miscadmin.h"
 #include "parser/parse_func.h"
 #include "rewrite/rewriteDefine.h"
+#include "storage/lmgr.h"
 #include "tcop/utility.h"
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
@@ -945,7 +946,9 @@ AlterObjectOwner_internal(Relation rel, Oid objectId, Oid new_ownerId)
        Oid                     old_ownerId;
        Oid                     namespaceId = InvalidOid;
 
-       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));
@@ -1044,6 +1047,8 @@ AlterObjectOwner_internal(Relation rel, Oid objectId, Oid new_ownerId)
                /* Perform actual update */
                CatalogTupleUpdate(rel, &newtup->t_self, newtup);
 
+               UnlockTuple(rel, &oldtup->t_self, InplaceUpdateTupleLock);
+
                /*
                 * Update owner dependency reference.  When working on a large object,
                 * we have to translate back to the OID conventionally used for LOs'
@@ -1061,6 +1066,8 @@ AlterObjectOwner_internal(Relation rel, Oid objectId, Oid new_ownerId)
        }
        else
        {
+               UnlockTuple(rel, &oldtup->t_self, InplaceUpdateTupleLock);
+
                /*
                 * No need to change anything.  But when working on a large object, we
                 * have to translate back to the OID conventionally used for LOs'

diff --git a/src/include/catalog/objectaddress.h b/src/include/catalog/objectaddress.h
index 2b4e104bb9d64b79be2260f73ab18b1ff4a76b2e..3903e7a6f2e3b06b38c86dc98703628df00eb4f1 100644 (file)
--- 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
index 6bc935c14edd4f1733d43b8f333d646542ac9be6..8c4bdc2d3493af349d7a6821d80b56673aafecb0 100644 (file)
--- a/src/test/regress/expected/database.out
+++ b/src/test/regress/expected/database.out
@@ -11,4 +11,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
index dbb899c41cecbd8adbe9d3fbe7491da919a2ee75..edd62128268c57965807ce63a2a0e854fdb3d18b 100644 (file)
--- a/src/test/regress/sql/database.sql
+++ b/src/test/regress/sql/database.sql
@@ -13,4 +13,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;