wip_idxlockmode_and_lock_upgrade_fix.patch

diff --git a/src/backend/executor/nodeIndexonlyscan.c b/src/backend/executor/nodeIndexonlyscan.c
index d1201a1807..1318904135 100644
--- a/src/backend/executor/nodeIndexonlyscan.c
+++ b/src/backend/executor/nodeIndexonlyscan.c
@@ -493,7 +493,7 @@ ExecInitIndexOnlyScan(IndexOnlyScan *node, EState *estate, int eflags)
 {
 	IndexOnlyScanState *indexstate;
 	Relation	currentRelation;
-	bool		relistarget;
+	LOCKMODE	lockmode;
 	TupleDesc	tupDesc;
 
 	/*
@@ -555,16 +555,9 @@ ExecInitIndexOnlyScan(IndexOnlyScan *node, EState *estate, int eflags)
 	if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
 		return indexstate;
 
-	/*
-	 * Open the index relation.
-	 *
-	 * If the parent table is one of the target relations of the query, then
-	 * InitPlan already opened and write-locked the index, so we can avoid
-	 * taking another lock here.  Otherwise we need a normal reader's lock.
-	 */
-	relistarget = ExecRelationIsTargetRelation(estate, node->scan.scanrelid);
-	indexstate->ioss_RelationDesc = index_open(node->indexid,
-											   relistarget ? NoLock : AccessShareLock);
+	/* Open the index relation. */
+	lockmode = exec_rt_fetch(node->scan.scanrelid, estate)->idxlockmode;
+	indexstate->ioss_RelationDesc = index_open(node->indexid, lockmode);
 
 	/*
 	 * Initialize index-specific scan state
diff --git a/src/backend/executor/nodeIndexscan.c b/src/backend/executor/nodeIndexscan.c
index f209173a85..fa5b587769 100644
--- a/src/backend/executor/nodeIndexscan.c
+++ b/src/backend/executor/nodeIndexscan.c
@@ -920,7 +920,7 @@ ExecInitIndexScan(IndexScan *node, EState *estate, int eflags)
 {
 	IndexScanState *indexstate;
 	Relation	currentRelation;
-	bool		relistarget;
+	LOCKMODE	lockmode;
 
 	/*
 	 * create state structure
@@ -983,16 +983,9 @@ ExecInitIndexScan(IndexScan *node, EState *estate, int eflags)
 	if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
 		return indexstate;
 
-	/*
-	 * Open the index relation.
-	 *
-	 * If the parent table is one of the target relations of the query, then
-	 * InitPlan already opened and write-locked the index, so we can avoid
-	 * taking another lock here.  Otherwise we need a normal reader's lock.
-	 */
-	relistarget = ExecRelationIsTargetRelation(estate, node->scan.scanrelid);
-	indexstate->iss_RelationDesc = index_open(node->indexid,
-											  relistarget ? NoLock : AccessShareLock);
+	/* Open the index relation. */
+	lockmode = exec_rt_fetch(node->scan.scanrelid, estate)->idxlockmode;
+	indexstate->iss_RelationDesc = index_open(node->indexid, lockmode);
 
 	/*
 	 * Initialize index-specific scan state
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index db49968409..6c4547fcbd 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2354,6 +2354,7 @@ _copyRangeTblEntry(const RangeTblEntry *from)
 	COPY_SCALAR_FIELD(relid);
 	COPY_SCALAR_FIELD(relkind);
 	COPY_SCALAR_FIELD(rellockmode);
+	COPY_SCALAR_FIELD(idxlockmode);
 	COPY_NODE_FIELD(tablesample);
 	COPY_NODE_FIELD(subquery);
 	COPY_SCALAR_FIELD(security_barrier);
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 3a084b4d1f..a8dfc390b3 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2630,6 +2630,7 @@ _equalRangeTblEntry(const RangeTblEntry *a, const RangeTblEntry *b)
 	COMPARE_SCALAR_FIELD(relid);
 	COMPARE_SCALAR_FIELD(relkind);
 	COMPARE_SCALAR_FIELD(rellockmode);
+	COMPARE_SCALAR_FIELD(idxlockmode);
 	COMPARE_NODE_FIELD(tablesample);
 	COMPARE_NODE_FIELD(subquery);
 	COMPARE_SCALAR_FIELD(security_barrier);
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index f0c396530d..7ce6f4b976 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -3127,6 +3127,7 @@ _outRangeTblEntry(StringInfo str, const RangeTblEntry *node)
 			WRITE_OID_FIELD(relid);
 			WRITE_CHAR_FIELD(relkind);
 			WRITE_INT_FIELD(rellockmode);
+			WRITE_INT_FIELD(idxlockmode);
 			WRITE_NODE_FIELD(tablesample);
 			break;
 		case RTE_SUBQUERY:
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index e117867de5..74895f256c 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -1362,6 +1362,7 @@ _readRangeTblEntry(void)
 			READ_OID_FIELD(relid);
 			READ_CHAR_FIELD(relkind);
 			READ_INT_FIELD(rellockmode);
+			READ_INT_FIELD(idxlockmode);
 			READ_NODE_FIELD(tablesample);
 			break;
 		case RTE_SUBQUERY:
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index c729a99f8b..6369c2caa6 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -122,6 +122,7 @@ typedef struct
 } WindowClauseSortData;
 
 /* Local functions */
+static void finalize_lockmodes(PlannedStmt *stmt);
 static Node *preprocess_expression(PlannerInfo *root, Node *expr, int kind);
 static void preprocess_qual_conditions(PlannerInfo *root, Node *jtnode);
 static void inheritance_planner(PlannerInfo *root);
@@ -565,9 +566,155 @@ standard_planner(Query *parse, int cursorOptions, ParamListInfo boundParams)
 			result->jitFlags |= PGJIT_DEFORM;
 	}
 
+	/*
+	 * Determine correct lock modes for each rtable entry and the indexes
+	 * belonging to it.
+	 */
+	finalize_lockmodes(result);
+
 	return result;
 }
 
+typedef struct RelationLockmodeElem
+{
+	Oid			relid;		/* hash key -- must be first */
+	LOCKMODE	lockmode;
+} RelationLockmodeElem;
+
+
+/*
+ * finalize_lockmodes
+ *		Process stmt's rtable and determine the strongest lock level for each
+ *		relation and upgrade weaker locks to the strongest lock level for that
+ *		relation.  Also determine the lock level require for each relation's
+ *		index.
+ */
+static void
+finalize_lockmodes(PlannedStmt *stmt)
+{
+	Bitmapset  *resultRelids = NULL;
+	List	   *rtable = stmt->rtable;
+	ListCell   *lc;
+	int			relid;
+
+	/*
+	 * Determine the strongest lock level for each relation in rtable.  We
+	 * must apply the strongest lock of each relation if the same relation is
+	 * seen more than once where the lock levels vary.  This works around lock
+	 * upgrade hazards we might see if we obtained the weaker lock followed by
+	 * the stronger lock level.  We need only attempt this when there are
+	 * multiple entries in the rtable.
+	 */
+	if (list_length(rtable) > 1)
+	{
+		RelationLockmodeElem   *elem;
+		HTAB	   *htab;
+		HASHCTL		ctl;
+		Oid			relid;
+		bool		found;
+		bool		applystrongest = false;
+
+		memset(&ctl, 0, sizeof(ctl));
+		ctl.keysize = sizeof(Oid);
+		ctl.entrysize = sizeof(RelationLockmodeElem);
+		ctl.hcxt = CurrentMemoryContext;
+
+		htab = hash_create("Lockmode table", list_length(rtable),
+			&ctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+
+		foreach(lc, rtable)
+		{
+			RangeTblEntry *rte = lfirst(lc);
+
+			if (rte->relkind != RELKIND_RELATION)
+				continue;
+
+			relid = rte->relid;
+			elem = (RelationLockmodeElem *)
+				hash_search(htab, &relid, HASH_ENTER, &found);
+
+			/*
+			 * When we've seen this relation before and the lockmode varies
+			 * from the last time we saw this rel, then mark that we need to
+			 * make another pass over the list to apply the strongest of the
+			 * seen lock modes.
+			 */
+			if (found)
+			{
+				if (elem->lockmode != rte->rellockmode)
+				{
+					applystrongest = true;
+					elem->lockmode = Max(elem->lockmode, rte->rellockmode);
+				}
+			}
+			else
+				elem->lockmode = rte->rellockmode;
+		}
+
+		if (applystrongest)
+		{
+			foreach(lc, rtable)
+			{
+				RangeTblEntry *rte = lfirst(lc);
+
+				if (rte->relkind != RELKIND_RELATION)
+					continue;
+
+				relid = rte->relid;
+
+				elem = (RelationLockmodeElem *)
+					hash_search(htab, &relid, HASH_FIND, &found);
+				Assert(found);
+
+				rte->rellockmode = elem->lockmode;
+			}
+		}
+
+		hash_destroy(htab);
+	}
+
+	if (stmt->commandType == CMD_INSERT || stmt->commandType == CMD_UPDATE)
+	{
+		foreach(lc, stmt->resultRelations)
+		{
+			relid = lfirst_int(lc);
+
+			resultRelids = bms_add_member(resultRelids, relid);
+		}
+	}
+
+	/* Determine index lock mode */
+	relid = 1;
+	foreach(lc, rtable)
+	{
+		RangeTblEntry *rte = lfirst(lc);
+
+		if (rte->relkind != RELKIND_RELATION)
+		{
+			relid++;
+			continue;
+		}
+
+		/*
+		 * SELECT always only will require an AccessShareLock on the indexes,
+		 * DELETE is the same since we're not modifying the index, only
+		 * marking the tuple on the heap as dead.
+		 */
+		if (stmt->commandType == CMD_SELECT || stmt->commandType == CMD_DELETE)
+			rte->idxlockmode = AccessShareLock;
+
+		/*
+		 * For INSERT AND UPDATE we require a RowExclusiveLock only for result
+		 * relations. resultRelids will be NULL when not an INSERT or UPDATE.
+		 */
+		else if (bms_is_member(relid, resultRelids))
+			rte->idxlockmode = RowExclusiveLock;
+		else
+			rte->idxlockmode = AccessShareLock; /* XXX how about utility commands? */
+
+		relid++;
+	}
+}
 
 /*--------------------
  * subquery_planner
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index e5bdc1cec5..6dbda5645d 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -987,6 +987,7 @@ typedef struct RangeTblEntry
 	Oid			relid;			/* OID of the relation */
 	char		relkind;		/* relation kind (see pg_class.relkind) */
 	int			rellockmode;	/* lock level that query requires on the rel */
+	int			idxlockmode;	/* lock level required for rel's indexes */
 	struct TableSampleClause *tablesample;	/* sampling info, or NULL */
 
 	/*