Proper object locking for GRANT/REVOKE

Refactor objectNamesToOids() to use get_object_address() internally if
possible.  Not only does this save a lot of code, it also allows us to
use the object locking provided by get_object_address() for
GRANT/REVOKE.  There was previously a code comment that complained
about the lack of locking in objectNamesToOids(), which is now fixed.

The check in ExecGrant_Type_check() is obsolete because
get_object_address_type() already does the same check.

Reviewed-by: Bertrand Drouvot <[email protected]>
Discussion: https://www.postgresql.org/message-id/flat/bf72b82c-124d-4efa-a484-bb928e9494e4@eisentraut.org

diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index 95eb0b12277b11655cefecc24a928cd9f1a43c95..aaf96a965e42af821ee945bc355f7a4fe0e58c3a 100644 (file)
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -659,147 +659,77 @@ ExecGrantStmt_oids(InternalGrant *istmt)
  * objectNamesToOids
  *
  * Turn a list of object names of a given type into an Oid list.
- *
- * XXX: This function doesn't take any sort of locks on the objects whose
- * names it looks up.  In the face of concurrent DDL, we might easily latch
- * onto an old version of an object, causing the GRANT or REVOKE statement
- * to fail.
  */
 static List *
 objectNamesToOids(ObjectType objtype, List *objnames, bool is_grant)
 {
    List       *objects = NIL;
    ListCell   *cell;
+   const LOCKMODE lockmode = AccessShareLock;
 
    Assert(objnames != NIL);
 
    switch (objtype)
    {
-       case OBJECT_TABLE:
-       case OBJECT_SEQUENCE:
-           foreach(cell, objnames)
-           {
-               RangeVar   *relvar = (RangeVar *) lfirst(cell);
-               Oid         relOid;
-
-               relOid = RangeVarGetRelid(relvar, NoLock, false);
-               objects = lappend_oid(objects, relOid);
-           }
-           break;
-       case OBJECT_DATABASE:
-           foreach(cell, objnames)
-           {
-               char       *dbname = strVal(lfirst(cell));
-               Oid         dbid;
-
-               dbid = get_database_oid(dbname, false);
-               objects = lappend_oid(objects, dbid);
-           }
-           break;
-       case OBJECT_DOMAIN:
-       case OBJECT_TYPE:
-           foreach(cell, objnames)
-           {
-               List       *typname = (List *) lfirst(cell);
-               Oid         oid;
-
-               oid = typenameTypeId(NULL, makeTypeNameFromNameList(typname));
-               objects = lappend_oid(objects, oid);
-           }
-           break;
-       case OBJECT_FUNCTION:
-           foreach(cell, objnames)
-           {
-               ObjectWithArgs *func = (ObjectWithArgs *) lfirst(cell);
-               Oid         funcid;
+       default:
 
-               funcid = LookupFuncWithArgs(OBJECT_FUNCTION, func, false);
-               objects = lappend_oid(objects, funcid);
-           }
-           break;
-       case OBJECT_LANGUAGE:
+           /*
+            * For most object types, we use get_object_address() directly.
+            */
            foreach(cell, objnames)
            {
-               char       *langname = strVal(lfirst(cell));
-               Oid         oid;
+               ObjectAddress address;
 
-               oid = get_language_oid(langname, false);
-               objects = lappend_oid(objects, oid);
+               address = get_object_address(objtype, lfirst(cell), NULL, lockmode, false);
+               objects = lappend_oid(objects, address.objectId);
            }
            break;
-       case OBJECT_LARGEOBJECT:
-           foreach(cell, objnames)
-           {
-               Oid         lobjOid = oidparse(lfirst(cell));
 
-               if (!LargeObjectExists(lobjOid))
-                   ereport(ERROR,
-                           (errcode(ERRCODE_UNDEFINED_OBJECT),
-                            errmsg("large object %u does not exist",
-                                   lobjOid)));
-
-               objects = lappend_oid(objects, lobjOid);
-           }
-           break;
-       case OBJECT_SCHEMA:
-           foreach(cell, objnames)
-           {
-               char       *nspname = strVal(lfirst(cell));
-               Oid         oid;
+       case OBJECT_TABLE:
+       case OBJECT_SEQUENCE:
 
-               oid = get_namespace_oid(nspname, false);
-               objects = lappend_oid(objects, oid);
-           }
-           break;
-       case OBJECT_PROCEDURE:
+           /*
+            * Here, we don't use get_object_address().  It requires that the
+            * specified object type match the actual type of the object, but
+            * in GRANT/REVOKE, all table-like things are addressed as TABLE.
+            */
            foreach(cell, objnames)
            {
-               ObjectWithArgs *func = (ObjectWithArgs *) lfirst(cell);
-               Oid         procid;
+               RangeVar   *relvar = (RangeVar *) lfirst(cell);
+               Oid         relOid;
 
-               procid = LookupFuncWithArgs(OBJECT_PROCEDURE, func, false);
-               objects = lappend_oid(objects, procid);
+               relOid = RangeVarGetRelid(relvar, lockmode, false);
+               objects = lappend_oid(objects, relOid);
            }
            break;
-       case OBJECT_ROUTINE:
-           foreach(cell, objnames)
-           {
-               ObjectWithArgs *func = (ObjectWithArgs *) lfirst(cell);
-               Oid         routid;
 
-               routid = LookupFuncWithArgs(OBJECT_ROUTINE, func, false);
-               objects = lappend_oid(objects, routid);
-           }
-           break;
-       case OBJECT_TABLESPACE:
-           foreach(cell, objnames)
-           {
-               char       *spcname = strVal(lfirst(cell));
-               Oid         spcoid;
+       case OBJECT_DOMAIN:
+       case OBJECT_TYPE:
 
-               spcoid = get_tablespace_oid(spcname, false);
-               objects = lappend_oid(objects, spcoid);
-           }
-           break;
-       case OBJECT_FDW:
+           /*
+            * The parse representation of types and domains in privilege
+            * targets is different from that expected by get_object_address()
+            * (for parse conflict reasons), so we have to do a bit of
+            * conversion here.
+            */
            foreach(cell, objnames)
            {
-               char       *fdwname = strVal(lfirst(cell));
-               Oid         fdwid = get_foreign_data_wrapper_oid(fdwname, false);
+               List       *typname = (List *) lfirst(cell);
+               TypeName   *tn = makeTypeNameFromNameList(typname);
+               ObjectAddress address;
+               Relation    relation;
 
-               objects = lappend_oid(objects, fdwid);
+               address = get_object_address(objtype, (Node *) tn, &relation, lockmode, false);
+               Assert(relation == NULL);
+               objects = lappend_oid(objects, address.objectId);
            }
            break;
-       case OBJECT_FOREIGN_SERVER:
-           foreach(cell, objnames)
-           {
-               char       *srvname = strVal(lfirst(cell));
-               Oid         srvid = get_foreign_server_oid(srvname, false);
 
-               objects = lappend_oid(objects, srvid);
-           }
-           break;
        case OBJECT_PARAMETER_ACL:
+
+           /*
+            * Parameters are handled completely differently.
+            */
            foreach(cell, objnames)
            {
                /*
@@ -830,9 +760,6 @@ objectNamesToOids(ObjectType objtype, List *objnames, bool is_grant)
                    objects = lappend_oid(objects, parameterId);
            }
            break;
-       default:
-           elog(ERROR, "unrecognized GrantStmt.objtype: %d",
-                (int) objtype);
    }
 
    return objects;
@@ -2458,14 +2385,6 @@ ExecGrant_Type_check(InternalGrant *istmt, HeapTuple tuple)
                (errcode(ERRCODE_INVALID_GRANT_OPERATION),
                 errmsg("cannot set privileges of multirange types"),
                 errhint("Set the privileges of the range type instead.")));
-
-   /* Used GRANT DOMAIN on a non-domain? */
-   if (istmt->objtype == OBJECT_DOMAIN &&
-       pg_type_tuple->typtype != TYPTYPE_DOMAIN)
-       ereport(ERROR,
-               (errcode(ERRCODE_WRONG_OBJECT_TYPE),
-                errmsg("\"%s\" is not a domain",
-                       NameStr(pg_type_tuple->typname))));
 }
 
 static void

diff --git a/src/test/isolation/expected/intra-grant-inplace.out b/src/test/isolation/expected/intra-grant-inplace.out
index 4e9695a02146c0a15adf1e94a6093e6fe9ab70c6..1aa9da622da05a0628a1ea2eff36b42168dfbe3f 100644 (file)
--- a/src/test/isolation/expected/intra-grant-inplace.out
+++ b/src/test/isolation/expected/intra-grant-inplace.out
@@ -248,6 +248,6 @@ relhasindex
 -----------
 (0 rows)
 
-s4: WARNING:  got: cache lookup failed for relation REDACTED
+s4: WARNING:  got: relation "intra_grant_inplace" does not exist
 step revoke4: <... completed>
 step r3: ROLLBACK;