/*-------------------------------------------------------------------------

 *

 * lockcmds.c

 *    LOCK command support code

 *

 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994, Regents of the University of California

 *

 *

 * IDENTIFICATION

 *    src/backend/commands/lockcmds.c

 *

 *-------------------------------------------------------------------------

 */

#include "postgres.h"

#include "access/table.h"

#include "access/xact.h"

#include "catalog/namespace.h"

#include "catalog/pg_inherits.h"

#include "commands/lockcmds.h"

#include "miscadmin.h"

#include "nodes/nodeFuncs.h"

#include "rewrite/rewriteHandler.h"

#include "storage/lmgr.h"

#include "utils/acl.h"

#include "utils/lsyscache.h"

#include "utils/syscache.h"

static void LockTableRecurse(Oid reloid, LOCKMODE lockmode, bool nowait);

static AclResult LockTableAclCheck(Oid reloid, LOCKMODE lockmode, Oid userid);

static void RangeVarCallbackForLockTable(const RangeVar *rv, Oid relid,

                     Oid oldrelid, void *arg);

static void LockViewRecurse(Oid reloid, LOCKMODE lockmode, bool nowait,

              List *ancestor_views);

/*

 * LOCK TABLE

 */

void

LockTableCommand(LockStmt *lockstmt)

{

  ListCell   *p;

  /*

   * Iterate over the list and process the named relations one at a time

   */

  foreach(p, lockstmt->relations)

  {

    RangeVar   *rv = (RangeVar *) lfirst(p);

    bool    recurse = rv->inh;

    Oid     reloid;

    reloid = RangeVarGetRelidExtended(rv, lockstmt->mode,

                      lockstmt->nowait ? RVR_NOWAIT : 0,

                      RangeVarCallbackForLockTable,

                      &lockstmt->mode);

    if (get_rel_relkind(reloid) == RELKIND_VIEW)

      LockViewRecurse(reloid, lockstmt->mode, lockstmt->nowait, NIL);

    else if (recurse)

      LockTableRecurse(reloid, lockstmt->mode, lockstmt->nowait);

  }

}

/*

 * Before acquiring a table lock on the named table, check whether we have

 * permission to do so.

 */

static void

RangeVarCallbackForLockTable(const RangeVar *rv, Oid relid, Oid oldrelid,

               void *arg)

{

  LOCKMODE  lockmode = *(LOCKMODE *) arg;

  char    relkind;

  char    relpersistence;

  AclResult aclresult;

  if (!OidIsValid(relid))

    return;         /* doesn't exist, so no permissions check */

  relkind = get_rel_relkind(relid);

  if (!relkind)

    return;         /* woops, concurrently dropped; no permissions

                 * check */

  /*

   * Note: Conflict log tables are deliberately NOT blocked here, even

   * though other direct DDL on them is rejected elsewhere. pg_dump relies

   * on being able to take an ACCESS SHARE lock on these tables to safely

   * dump their definitions during a binary upgrade, so we permit LOCK

   * commands on them and treat them like ordinary tables here. It's true

   * that a strong lock (ShareLock or above) on such a table would conflict

   * with the RowExclusiveLock taken by the apply worker's inserts and could

   * stall conflict logging as well as the apply worker for as long as it is

   * held. But locking a system-managed conflict log table is an unusual

   * thing to do, and it doesn't seem worth the trouble of filtering by lock

   * mode here.

   */

  if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE &&

    relkind != RELKIND_VIEW)

    ereport(ERROR,

        (errcode(ERRCODE_WRONG_OBJECT_TYPE),

         errmsg("cannot lock relation \"%s\"",

            rv->relname),

         errdetail_relkind_not_supported(relkind)));

  /*

   * Make note if a temporary relation has been accessed in this

   * transaction.

   */

  relpersistence = get_rel_persistence(relid);

  if (relpersistence == RELPERSISTENCE_TEMP)

    MyXactFlags |= XACT_FLAGS_ACCESSEDTEMPNAMESPACE;

  /* Check permissions. */

  aclresult = LockTableAclCheck(relid, lockmode, GetUserId());

  if (aclresult != ACLCHECK_OK)

    aclcheck_error(aclresult, get_relkind_objtype(get_rel_relkind(relid)), rv->relname);

}

/*

 * Apply LOCK TABLE recursively over an inheritance tree

 *

 * This doesn't check permission to perform LOCK TABLE on the child tables,

 * because getting here means that the user has permission to lock the

 * parent which is enough.

 */

static void

LockTableRecurse(Oid reloid, LOCKMODE lockmode, bool nowait)

{

  List     *children;

  ListCell   *lc;

  children = find_all_inheritors(reloid, NoLock, NULL);

  foreach(lc, children)

  {

    Oid     childreloid = lfirst_oid(lc);

    /* Parent already locked. */

    if (childreloid == reloid)

      continue;

    if (!nowait)

      LockRelationOid(childreloid, lockmode);

    else if (!ConditionalLockRelationOid(childreloid, lockmode))

    {

      /* try to throw error by name; relation could be deleted... */

      char     *relname = get_rel_name(childreloid);

      if (!relname)

        continue;   /* child concurrently dropped, just skip it */

      ereport(ERROR,

          (errcode(ERRCODE_LOCK_NOT_AVAILABLE),

           errmsg("could not obtain lock on relation \"%s\"",

              relname)));

    }

    /*

     * Even if we got the lock, child might have been concurrently

     * dropped. If so, we can skip it.

     */

    if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(childreloid)))

    {

      /* Release useless lock */

      UnlockRelationOid(childreloid, lockmode);

      continue;

    }

  }

}

/*

 * Apply LOCK TABLE recursively over a view

 *

 * All tables and views appearing in the view definition query are locked

 * recursively with the same lock mode.

 */

typedef struct

{

  LOCKMODE  lockmode;   /* lock mode to use */

  bool    nowait;     /* no wait mode */

  Oid     check_as_user;  /* user for checking the privilege */

  Oid     viewoid;    /* OID of the view to be locked */

  List     *ancestor_views; /* OIDs of ancestor views */

} LockViewRecurse_context;

static bool

LockViewRecurse_walker(Node *node, LockViewRecurse_context *context)

{

  if (node == NULL)

    return false;

  if (IsA(node, Query))

  {

    Query    *query = (Query *) node;

    ListCell   *rtable;

    foreach(rtable, query->rtable)

    {

      RangeTblEntry *rte = lfirst(rtable);

      AclResult aclresult;

      Oid     relid = rte->relid;

      char    relkind = rte->relkind;

      char     *relname = get_rel_name(relid);

      /* Currently, we only allow plain tables or views to be locked. */

      if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE &&

        relkind != RELKIND_VIEW)

        continue;

      /*

       * Conflict log tables only support SELECT, DELETE, and TRUNCATE.

       * A direct LOCK on them is permitted solely so that pg_dump can

       * lock them during a binary upgrade; locking one indirectly by

       * locking a view over it is not needed for that, so skip it here

       * rather than locking it.

       */

      if (IsConflictLogTableNamespace(get_rel_namespace(relid)))

        continue;

      /*

       * We might be dealing with a self-referential view.  If so, we

       * can just stop recursing, since we already locked it.

       */

      if (list_member_oid(context->ancestor_views, relid))

        continue;

      /*

       * Check permissions as the specified user.  This will either be

       * the view owner or the current user.

       */

      aclresult = LockTableAclCheck(relid, context->lockmode,

                      context->check_as_user);

      if (aclresult != ACLCHECK_OK)

        aclcheck_error(aclresult, get_relkind_objtype(relkind), relname);

      /* We have enough rights to lock the relation; do so. */

      if (!context->nowait)

        LockRelationOid(relid, context->lockmode);

      else if (!ConditionalLockRelationOid(relid, context->lockmode))

        ereport(ERROR,

            (errcode(ERRCODE_LOCK_NOT_AVAILABLE),

             errmsg("could not obtain lock on relation \"%s\"",

                relname)));

      if (relkind == RELKIND_VIEW)

        LockViewRecurse(relid, context->lockmode, context->nowait,

                context->ancestor_views);

      else if (rte->inh)

        LockTableRecurse(relid, context->lockmode, context->nowait);

    }

    return query_tree_walker(query,

                 LockViewRecurse_walker,

                 context,

                 QTW_IGNORE_JOINALIASES);

  }

  return expression_tree_walker(node,

                  LockViewRecurse_walker,

                  context);

}

static void

LockViewRecurse(Oid reloid, LOCKMODE lockmode, bool nowait,

        List *ancestor_views)

{

  LockViewRecurse_context context;

  Relation  view;

  Query    *viewquery;

  /* caller has already locked the view */

  view = table_open(reloid, NoLock);

  viewquery = get_view_query(view);

  /*

   * If the view has the security_invoker property set, check permissions as

   * the current user.  Otherwise, check permissions as the view owner.

   */

  context.lockmode = lockmode;

  context.nowait = nowait;

  if (RelationHasSecurityInvoker(view))

    context.check_as_user = GetUserId();

  else

    context.check_as_user = view->rd_rel->relowner;

  context.viewoid = reloid;

  context.ancestor_views = lappend_oid(ancestor_views, reloid);

  LockViewRecurse_walker((Node *) viewquery, &context);

  context.ancestor_views = list_delete_last(context.ancestor_views);

  table_close(view, NoLock);

}

/*

 * Check whether the current user is permitted to lock this relation.

 */

static AclResult

LockTableAclCheck(Oid reloid, LOCKMODE lockmode, Oid userid)

{

  AclResult aclresult;

  AclMode   aclmask;

  /* any of these privileges permit any lock mode */

  aclmask = ACL_MAINTAIN | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE;

  /* SELECT privileges also permit ACCESS SHARE and below */

  if (lockmode <= AccessShareLock)

    aclmask |= ACL_SELECT;

  /* INSERT privileges also permit ROW EXCLUSIVE and below */

  if (lockmode <= RowExclusiveLock)

    aclmask |= ACL_INSERT;

  aclresult = pg_class_aclcheck(reloid, userid, aclmask);

  return aclresult;

}