Re: executor relation handling

I wrote:
> 1. You set up transformRuleStmt to insert AccessExclusiveLock into
> the "OLD" and "NEW" RTEs for a view.  This is surely wrong; we do
> not want to take exclusive lock on a view just to run a query using
> the view.  It should (usually, anyway) just be AccessShareLock.
> However, because addRangeTableEntryForRelation insists that you
> hold the requested lock type *now*, just changing the parameter
> to AccessShareLock doesn't work.
> I hacked around this for the moment by passing NoLock to
> addRangeTableEntryForRelation and then changing rte->lockmode
> after it returns, but man that's ugly.  It makes me wonder whether
> addRangeTableEntryForRelation should be checking the lockmode at all.

It occurred to me that it'd be reasonable to insist that the caller
holds a lock *at least as strong* as the one being recorded in the RTE,
and that there's also been discussions about verifying that some lock
is held when something like heap_open(foo, NoLock) is attempted.
So I dusted off the part of 0001 that did that, producing the
attached delta patch.

Unfortunately, I can't commit this, because it exposes at least two
pre-existing bugs :-(.  So we'll need to fix those first, which seems
like it should be a separate thread.  I'm just parking this here for
the moment.

I think that the call sites should ultimately look like

    Assert(CheckRelationLockedByMe(...));

but for hunting down the places where the assertion currently fails,
it's more convenient if it's just an elog(WARNING).

            regards, tom lane

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 3395445..f8a55ee 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -1137,6 +1137,11 @@ relation_open(Oid relationId, LOCKMODE lockmode)
     if (!RelationIsValid(r))
         elog(ERROR, "could not open relation with OID %u", relationId);

+    if (lockmode == NoLock &&
+        !CheckRelationLockedByMe(r, AccessShareLock, true))
+        elog(WARNING, "relation_open: no lock held on %s",
+             RelationGetRelationName(r));
+
     /* Make note that we've accessed a temporary relation */
     if (RelationUsesLocalBuffers(r))
         MyXactFlags |= XACT_FLAGS_ACCESSEDTEMPREL;
@@ -1183,6 +1188,11 @@ try_relation_open(Oid relationId, LOCKMODE lockmode)
     if (!RelationIsValid(r))
         elog(ERROR, "could not open relation with OID %u", relationId);

+    if (lockmode == NoLock &&
+        !CheckRelationLockedByMe(r, AccessShareLock, true))
+        elog(WARNING, "try_relation_open: no lock held on %s",
+             RelationGetRelationName(r));
+
     /* Make note that we've accessed a temporary relation */
     if (RelationUsesLocalBuffers(r))
         MyXactFlags |= XACT_FLAGS_ACCESSEDTEMPREL;
@@ -8084,6 +8094,10 @@ ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_changed, bool *

     idx_rel = RelationIdGetRelation(replidindex);

+    if (!CheckRelationLockedByMe(idx_rel, AccessShareLock, true))
+        elog(WARNING, "ExtractReplicaIdentity: no lock held on %s",
+             RelationGetRelationName(idx_rel));
+
     /* deform tuple, so we have fast access to columns */
     heap_deform_tuple(tp, desc, values, nulls);

diff --git a/src/backend/parser/parse_relation.c b/src/backend/parser/parse_relation.c
index da600dc..aaf5544 100644
--- a/src/backend/parser/parse_relation.c
+++ b/src/backend/parser/parse_relation.c
@@ -28,6 +28,7 @@
 #include "parser/parse_enr.h"
 #include "parser/parse_relation.h"
 #include "parser/parse_type.h"
+#include "storage/lmgr.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "utils/rel.h"
@@ -1272,7 +1273,7 @@ addRangeTableEntry(ParseState *pstate,
  *
  * lockmode is the lock type required for query execution; it must be one
  * of AccessShareLock, RowShareLock, or RowExclusiveLock depending on the
- * RTE's role within the query.  The caller should always hold that lock mode
+ * RTE's role within the query.  The caller must hold that lock mode
  * or a stronger one.
  *
  * Note: properly, lockmode should be declared LOCKMODE not int, but that
@@ -1295,6 +1296,9 @@ addRangeTableEntryForRelation(ParseState *pstate,
     Assert(lockmode == AccessShareLock ||
            lockmode == RowShareLock ||
            lockmode == RowExclusiveLock);
+    if (!CheckRelationLockedByMe(rel, lockmode, true))
+        elog(WARNING, "addRangeTableEntryForRelation: no lock held on %s",
+             RelationGetRelationName(rel));

     rte->rtekind = RTE_RELATION;
     rte->alias = alias;
diff --git a/src/backend/storage/lmgr/lmgr.c b/src/backend/storage/lmgr/lmgr.c
index dc0a439..517ff8e 100644
--- a/src/backend/storage/lmgr/lmgr.c
+++ b/src/backend/storage/lmgr/lmgr.c
@@ -288,6 +288,51 @@ UnlockRelation(Relation relation, LOCKMODE lockmode)
 }

 /*
+ *        CheckRelationLockedByMe
+ *
+ * Returns true if current transaction holds a lock on 'relation' of mode
+ * 'lockmode'.  If 'orstronger' is true, a stronger lockmode is also OK.
+ * ("Stronger" is defined as "numerically higher", which is a bit
+ * semantically dubious but is OK for the purposes we use this for.)
+ */
+bool
+CheckRelationLockedByMe(Relation relation, LOCKMODE lockmode, bool orstronger)
+{
+    LOCKTAG        tag;
+
+    SET_LOCKTAG_RELATION(tag,
+                         relation->rd_lockInfo.lockRelId.dbId,
+                         relation->rd_lockInfo.lockRelId.relId);
+
+    if (LockHeldByMe(&tag, lockmode))
+        return true;
+
+    if (orstronger)
+    {
+        LOCKMODE    slockmode;
+
+        for (slockmode = lockmode + 1;
+             slockmode <= AccessExclusiveLock;
+             slockmode++)
+        {
+            if (LockHeldByMe(&tag, slockmode))
+            {
+#ifdef NOT_USED
+                /* Sometimes this might be useful for debugging purposes */
+                elog(WARNING, "lock mode %s substituted for %s on relation %s",
+                     GetLockmodeName(tag.locktag_lockmethodid, slockmode),
+                     GetLockmodeName(tag.locktag_lockmethodid, lockmode),
+                     RelationGetRelationName(relation));
+#endif
+                return true;
+            }
+        }
+    }
+
+    return false;
+}
+
+/*
  *        LockHasWaitersRelation
  *
  * This is a function to check whether someone else is waiting for a
diff --git a/src/backend/storage/lmgr/lock.c b/src/backend/storage/lmgr/lock.c
index 831ae62..10f6f60 100644
--- a/src/backend/storage/lmgr/lock.c
+++ b/src/backend/storage/lmgr/lock.c
@@ -564,6 +564,30 @@ DoLockModesConflict(LOCKMODE mode1, LOCKMODE mode2)
 }

 /*
+ * LockHeldByMe -- test whether lock 'locktag' is held with mode 'lockmode'
+ *        by the current transaction
+ */
+bool
+LockHeldByMe(const LOCKTAG *locktag, LOCKMODE lockmode)
+{
+    LOCALLOCKTAG localtag;
+    LOCALLOCK  *locallock;
+
+    /*
+     * See if there is a LOCALLOCK entry for this lock and lockmode
+     */
+    MemSet(&localtag, 0, sizeof(localtag)); /* must clear padding */
+    localtag.lock = *locktag;
+    localtag.mode = lockmode;
+
+    locallock = (LOCALLOCK *) hash_search(LockMethodLocalHash,
+                                          (void *) &localtag,
+                                          HASH_FIND, NULL);
+
+    return (locallock && locallock->nLocks > 0);
+}
+
+/*
  * LockHasWaiters -- look up 'locktag' and check if releasing this
  *        lock would wake up other processes waiting for it.
  */
diff --git a/src/include/storage/lmgr.h b/src/include/storage/lmgr.h
index a217de9..e5356b7 100644
--- a/src/include/storage/lmgr.h
+++ b/src/include/storage/lmgr.h
@@ -45,6 +45,8 @@ extern void UnlockRelationOid(Oid relid, LOCKMODE lockmode);
 extern void LockRelation(Relation relation, LOCKMODE lockmode);
 extern bool ConditionalLockRelation(Relation relation, LOCKMODE lockmode);
 extern void UnlockRelation(Relation relation, LOCKMODE lockmode);
+extern bool CheckRelationLockedByMe(Relation relation, LOCKMODE lockmode,
+                        bool orstronger);
 extern bool LockHasWaitersRelation(Relation relation, LOCKMODE lockmode);

 extern void LockRelationIdForSession(LockRelId *relid, LOCKMODE lockmode);
diff --git a/src/include/storage/lock.h b/src/include/storage/lock.h
index ff4df7f..a37fda7 100644
--- a/src/include/storage/lock.h
+++ b/src/include/storage/lock.h
@@ -540,6 +540,7 @@ extern void LockReleaseAll(LOCKMETHODID lockmethodid, bool allLocks);
 extern void LockReleaseSession(LOCKMETHODID lockmethodid);
 extern void LockReleaseCurrentOwner(LOCALLOCK **locallocks, int nlocks);
 extern void LockReassignCurrentOwner(LOCALLOCK **locallocks, int nlocks);
+extern bool LockHeldByMe(const LOCKTAG *locktag, LOCKMODE lockmode);
 extern bool LockHasWaiters(const LOCKTAG *locktag,
                LOCKMODE lockmode, bool sessionLock);
 extern VirtualTransactionId *GetLockConflicts(const LOCKTAG *locktag,