From 02874424d9b142981c113ac09a2d8fdab4b7d903 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Tue, 9 Jun 2026 21:45:37 +0200
Subject: [PATCH v1] FK-RI: Add heavy-weight locking option for RI checks

This implements RI checks that don't need row locks, and therefore
avoid MXact churn on frequently-referenced rows.

(Bump catalog version)
---
 doc/src/sgml/ref/alter_table.sgml             |   4 +-
 doc/src/sgml/ref/create_table.sgml            |  29 +++-
 src/backend/catalog/heap.c                    |   2 +
 src/backend/catalog/index.c                   |   1 +
 src/backend/catalog/pg_constraint.c           |   2 +
 src/backend/commands/tablecmds.c              |   8 +-
 src/backend/commands/trigger.c                |   1 +
 src/backend/commands/typecmds.c               |   2 +
 src/backend/parser/gram.y                     |  33 ++--
 src/backend/utils/adt/ri_triggers.c           | 160 +++++++++++++++---
 src/backend/utils/adt/ruleutils.c             |   3 +
 src/include/catalog/pg_constraint.h           |   2 +
 src/include/nodes/parsenodes.h                |   1 +
 .../fk-lock-key-index-no-conflicts.out        | 131 ++++++++++++++
 .../fk-lock-key-index-with-conflicts.out      | 125 ++++++++++++++
 src/test/isolation/isolation_schedule         |   2 +
 .../specs/fk-lock-key-index-no-conflicts.spec |  57 +++++++
 .../fk-lock-key-index-with-conflicts.spec     |  56 ++++++
 18 files changed, 578 insertions(+), 41 deletions(-)
 create mode 100644 src/test/isolation/expected/fk-lock-key-index-no-conflicts.out
 create mode 100644 src/test/isolation/expected/fk-lock-key-index-with-conflicts.out
 create mode 100644 src/test/isolation/specs/fk-lock-key-index-no-conflicts.spec
 create mode 100644 src/test/isolation/specs/fk-lock-key-index-with-conflicts.spec

diff --git a/doc/src/sgml/ref/alter_table.sgml b/doc/src/sgml/ref/alter_table.sgml
index dec34337d1a..4054e0db616 100644
--- a/doc/src/sgml/ref/alter_table.sgml
+++ b/doc/src/sgml/ref/alter_table.sgml
@@ -113,7 +113,7 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
   GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY [ ( <replaceable>sequence_options</replaceable> ) ] |
   UNIQUE [ NULLS [ NOT ] DISTINCT ] <replaceable class="parameter">index_parameters</replaceable> |
   PRIMARY KEY <replaceable class="parameter">index_parameters</replaceable> |
-  REFERENCES <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> ) ] [ MATCH FULL | MATCH PARTIAL | MATCH SIMPLE ]
+  REFERENCES <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> ) ] [ LOCK ROWS | LOCK KEY INDEX ] [ MATCH FULL | MATCH PARTIAL | MATCH SIMPLE ]
     [ ON DELETE <replaceable class="parameter">referential_action</replaceable> ] [ ON UPDATE <replaceable class="parameter">referential_action</replaceable> ] }
 [ DEFERRABLE | NOT DEFERRABLE ] [ INITIALLY DEFERRED | INITIALLY IMMEDIATE ] [ ENFORCED | NOT ENFORCED ]
 
@@ -126,7 +126,7 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
   PRIMARY KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] [, <replaceable class="parameter">column_name</replaceable> WITHOUT OVERLAPS ] ) <replaceable class="parameter">index_parameters</replaceable> |
   EXCLUDE [ USING <replaceable class="parameter">index_method</replaceable> ] ( <replaceable class="parameter">exclude_element</replaceable> WITH <replaceable class="parameter">operator</replaceable> [, ... ] ) <replaceable class="parameter">index_parameters</replaceable> [ WHERE ( <replaceable class="parameter">predicate</replaceable> ) ] |
   FOREIGN KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] [, PERIOD <replaceable class="parameter">column_name</replaceable> ] ) REFERENCES <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> [, ... ]  [, PERIOD <replaceable class="parameter">refcolumn</replaceable> ] ) ]
-    [ MATCH FULL | MATCH PARTIAL | MATCH SIMPLE ] [ ON DELETE <replaceable class="parameter">referential_action</replaceable> ] [ ON UPDATE <replaceable class="parameter">referential_action</replaceable> ] }
+    [ LOCK ROWS | LOCK KEY INDEX ] [ MATCH FULL | MATCH PARTIAL | MATCH SIMPLE ] [ ON DELETE <replaceable class="parameter">referential_action</replaceable> ] [ ON UPDATE <replaceable class="parameter">referential_action</replaceable> ] }
 [ DEFERRABLE | NOT DEFERRABLE ] [ INITIALLY DEFERRED | INITIALLY IMMEDIATE ] [ ENFORCED | NOT ENFORCED ]
 
 <phrase>and <replaceable class="parameter">table_constraint_using_index</replaceable> is:</phrase>
diff --git a/doc/src/sgml/ref/create_table.sgml b/doc/src/sgml/ref/create_table.sgml
index e342585c7f0..bc9924552e4 100644
--- a/doc/src/sgml/ref/create_table.sgml
+++ b/doc/src/sgml/ref/create_table.sgml
@@ -81,7 +81,7 @@ COMPRESSION <replaceable>compression_method</replaceable>
   GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY [ ( <replaceable>sequence_options</replaceable> ) ] |
   UNIQUE [ NULLS [ NOT ] DISTINCT ] <replaceable class="parameter">index_parameters</replaceable> |
   PRIMARY KEY <replaceable class="parameter">index_parameters</replaceable> |
-  REFERENCES <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> ) ] [ MATCH FULL | MATCH PARTIAL | MATCH SIMPLE ]
+  REFERENCES <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> ) ] [ LOCK ROWS | LOCK KEY INDEX ] [ MATCH FULL | MATCH PARTIAL | MATCH SIMPLE ]
     [ ON DELETE <replaceable class="parameter">referential_action</replaceable> ] [ ON UPDATE <replaceable class="parameter">referential_action</replaceable> ] }
 [ DEFERRABLE | NOT DEFERRABLE ] [ INITIALLY DEFERRED | INITIALLY IMMEDIATE ] [ ENFORCED | NOT ENFORCED ]
 
@@ -94,7 +94,7 @@ COMPRESSION <replaceable>compression_method</replaceable>
   PRIMARY KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] [, <replaceable class="parameter">column_name</replaceable> WITHOUT OVERLAPS ] ) <replaceable class="parameter">index_parameters</replaceable> |
   EXCLUDE [ USING <replaceable class="parameter">index_method</replaceable> ] ( <replaceable class="parameter">exclude_element</replaceable> WITH <replaceable class="parameter">operator</replaceable> [, ... ] ) <replaceable class="parameter">index_parameters</replaceable> [ WHERE ( <replaceable class="parameter">predicate</replaceable> ) ] |
   FOREIGN KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] [, PERIOD <replaceable class="parameter">column_name</replaceable> ] ) REFERENCES <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> [, ... ] [, PERIOD <replaceable class="parameter">refcolumn</replaceable> ] ) ]
-    [ MATCH FULL | MATCH PARTIAL | MATCH SIMPLE ] [ ON DELETE <replaceable
+    [ LOCK ROWS | LOCK KEY INDEX ] [ MATCH FULL | MATCH PARTIAL | MATCH SIMPLE ] [ ON DELETE <replaceable
 class="parameter">referential_action</replaceable> ] [ ON UPDATE <replaceable class="parameter">referential_action</replaceable> ] }
 [ DEFERRABLE | NOT DEFERRABLE ] [ INITIALLY DEFERRED | INITIALLY IMMEDIATE ] [ ENFORCED | NOT ENFORCED ]
 
@@ -1207,6 +1207,7 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
 
    <term><literal>FOREIGN KEY ( <replaceable class="parameter">column_name</replaceable> [, ... ] [, PERIOD <replaceable class="parameter">column_name</replaceable> ] )
     REFERENCES <replaceable class="parameter">reftable</replaceable> [ ( <replaceable class="parameter">refcolumn</replaceable> [, ... ] [, PERIOD <replaceable class="parameter">refcolumn</replaceable> ] ) ]
+    [ LOCK <replaceable class="parameter">locked_object</replaceable> ]
     [ MATCH <replaceable class="parameter">matchtype</replaceable> ]
     [ ON DELETE <replaceable class="parameter">referential_action</replaceable> ]
     [ ON UPDATE <replaceable class="parameter">referential_action</replaceable> ]</literal>
@@ -1263,6 +1264,30 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM
       tables and permanent tables.
      </para>
 
+     <para>
+      When a new row is inserted, or key values are modified in the
+      referencing table, we must protect the transaction against concurrent
+      removal of the value (and new insertions of removed values of the base
+      table) through locking.  There are two locking approaches available:
+      <literal>LOCK ROWS</literal> (which is the default), and <literal>LOCK
+      KEY INDEX</literal>.  <literal>LOCK ROWS</literal> locks the rows of
+      the base table with row-level <literal>FOR KEY SHARE</literal> locks
+      when the referencing table newly references those rows, preventing
+      those key values from being removed by a concurrent transaction.
+      <literal>LOCK ROWS</literal> allows for the high concurrency when key
+      values are frequently modified.  Instead of row-level locks, the
+      <literal>LOCK KEY INDEX</literal> foreign key locking option takes a
+      more heavy-handed approach; it requires a <literal>SHARE ROW EXCLUSIVE
+      </literal> lock on the index that backs the constraint when a key
+      value is removed or modified in the base table; and uses a <literal>ROW
+      EXCLUSIVE</literal> lock on the same index when new rows are inserted
+      into the referencing table.  The <literal>LOCK KEY INDEX</literal> is
+      therefore useful for foreign keys that see high amounts of insertions
+      and updates in the referencing table, but where these reference very
+      few and rarely updated rows in the base table, and when these key
+      updates in the base table never need to happen concurrently.
+     </para>
+
      <para>
       A value inserted into the referencing column(s) is matched against the
       values of the referenced table and referenced columns using the
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 88087654de9..d4d2c74c4fb 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -2248,6 +2248,7 @@ StoreRelCheck(Relation rel, const char *ccname, Node *expr,
 							  0,
 							  ' ',
 							  ' ',
+							  false,
 							  NULL,
 							  0,
 							  ' ',
@@ -2302,6 +2303,7 @@ StoreRelNotNull(Relation rel, const char *nnname, AttrNumber attnum,
 							  0,
 							  ' ',
 							  ' ',
+							  false,
 							  NULL,
 							  0,
 							  ' ',
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 9407c357f27..6c826b950f8 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -1994,6 +1994,7 @@ index_constraint_create(Relation heapRelation,
 								   0,
 								   ' ',
 								   ' ',
+								   false,
 								   NULL,
 								   0,
 								   ' ',
diff --git a/src/backend/catalog/pg_constraint.c b/src/backend/catalog/pg_constraint.c
index b12765ae691..bc117d7b52b 100644
--- a/src/backend/catalog/pg_constraint.c
+++ b/src/backend/catalog/pg_constraint.c
@@ -70,6 +70,7 @@ CreateConstraintEntry(const char *constraintName,
 					  int foreignNKeys,
 					  char foreignUpdateType,
 					  char foreignDeleteType,
+					  bool foreignLockKeyIndex,
 					  const int16 *fkDeleteSetCols,
 					  int numFkDeleteSetCols,
 					  char foreignMatchType,
@@ -200,6 +201,7 @@ CreateConstraintEntry(const char *constraintName,
 	values[Anum_pg_constraint_confupdtype - 1] = CharGetDatum(foreignUpdateType);
 	values[Anum_pg_constraint_confdeltype - 1] = CharGetDatum(foreignDeleteType);
 	values[Anum_pg_constraint_confmatchtype - 1] = CharGetDatum(foreignMatchType);
+	values[Anum_pg_constraint_conflockkeyindex - 1] = BoolGetDatum(foreignLockKeyIndex);
 	values[Anum_pg_constraint_conislocal - 1] = BoolGetDatum(conIsLocal);
 	values[Anum_pg_constraint_coninhcount - 1] = Int16GetDatum(conInhCount);
 	values[Anum_pg_constraint_connoinherit - 1] = BoolGetDatum(conNoInherit);
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index a33e22e8e61..f50a926ecdd 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -10884,6 +10884,7 @@ addFkConstraint(addFkConstraintSides fkside,
 									  numfks,
 									  fkconstraint->fk_upd_action,
 									  fkconstraint->fk_del_action,
+									  fkconstraint->fk_lockkeyindex,
 									  fkdelsetcols,
 									  numfkdelsetcols,
 									  fkconstraint->fk_matchtype,
@@ -11436,6 +11437,7 @@ CloneFkReferenced(Relation parentRel, Relation partitionRel)
 		fkconstraint->fk_matchtype = constrForm->confmatchtype;
 		fkconstraint->fk_upd_action = constrForm->confupdtype;
 		fkconstraint->fk_del_action = constrForm->confdeltype;
+		fkconstraint->fk_lockkeyindex = constrForm->conflockkeyindex;
 		fkconstraint->fk_del_set_cols = NIL;
 		fkconstraint->old_conpfeqop = NIL;
 		fkconstraint->old_pktable_oid = InvalidOid;
@@ -11699,6 +11701,7 @@ CloneFkReferencing(List **wqueue, Relation parentRel, Relation partRel)
 		fkconstraint->fk_matchtype = constrForm->confmatchtype;
 		fkconstraint->fk_upd_action = constrForm->confupdtype;
 		fkconstraint->fk_del_action = constrForm->confdeltype;
+		fkconstraint->fk_lockkeyindex = constrForm->conflockkeyindex;
 		fkconstraint->fk_del_set_cols = NIL;
 		fkconstraint->old_conpfeqop = NIL;
 		fkconstraint->old_pktable_oid = InvalidOid;
@@ -11839,7 +11842,8 @@ tryAttachPartitionForeignKey(List **wqueue,
 		partConstr->condeferred != parentConstr->condeferred ||
 		partConstr->confupdtype != parentConstr->confupdtype ||
 		partConstr->confdeltype != parentConstr->confdeltype ||
-		partConstr->confmatchtype != parentConstr->confmatchtype)
+		partConstr->confmatchtype != parentConstr->confmatchtype ||
+		partConstr->conflockkeyindex != parentConstr->conflockkeyindex)
 	{
 		ReleaseSysCache(parentConstrTup);
 		ReleaseSysCache(partcontup);
@@ -12567,6 +12571,7 @@ ATExecAlterFKConstrEnforceability(List **wqueue, ATAlterConstraint *cmdcon,
 		fkconstraint->fk_matchtype = currcon->confmatchtype;
 		fkconstraint->fk_upd_action = currcon->confupdtype;
 		fkconstraint->fk_del_action = currcon->confdeltype;
+		fkconstraint->fk_lockkeyindex = currcon->conflockkeyindex;
 		fkconstraint->deferrable = currcon->condeferrable;
 		fkconstraint->initdeferred = currcon->condeferred;
 
@@ -21547,6 +21552,7 @@ DetachPartitionFinalize(Relation rel, Relation partRel, bool concurrent,
 			fkconstraint->fk_matchtype = conform->confmatchtype;
 			fkconstraint->fk_upd_action = conform->confupdtype;
 			fkconstraint->fk_del_action = conform->confdeltype;
+			fkconstraint->fk_lockkeyindex = conform->conflockkeyindex;
 			fkconstraint->fk_del_set_cols = NIL;
 			fkconstraint->old_conpfeqop = NIL;
 			fkconstraint->old_pktable_oid = InvalidOid;
diff --git a/src/backend/commands/trigger.c b/src/backend/commands/trigger.c
index b87b4b40d07..5239a94b64c 100644
--- a/src/backend/commands/trigger.c
+++ b/src/backend/commands/trigger.c
@@ -831,6 +831,7 @@ CreateTriggerFiringOn(const CreateTrigStmt *stmt, const char *queryString,
 											  0,
 											  ' ',
 											  ' ',
+											  false,
 											  NULL,
 											  0,
 											  ' ',
diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index c4c3cdb5461..f69aa5227f7 100644
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -3659,6 +3659,7 @@ domainAddCheckConstraint(Oid domainOid, Oid domainNamespace, Oid baseTypeOid,
 							  0,
 							  ' ',
 							  ' ',
+							  false,
 							  NULL,
 							  0,
 							  ' ',
@@ -3767,6 +3768,7 @@ domainAddNotNullConstraint(Oid domainOid, Oid domainNamespace, Oid baseTypeOid,
 							  0,
 							  ' ',
 							  ' ',
+							  false,
 							  NULL,
 							  0,
 							  ' ',
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index ff4e1388c55..11a214d671b 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -610,6 +610,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 %type <list>	ColQualList
 %type <node>	ColConstraint ColConstraintElem ConstraintAttr
 %type <ival>	key_match
+%type <boolean> fklockclause opt_fklockclause
 %type <keyaction> key_delete key_update key_action
 %type <keyactions> key_actions
 %type <ival>	ConstraintAttributeSpec ConstraintAttributeElem
@@ -4221,7 +4222,7 @@ ColConstraintElem:
 
 					$$ = (Node *) n;
 				}
-			| REFERENCES qualified_name opt_column_list key_match key_actions
+			| REFERENCES qualified_name opt_column_list opt_fklockclause key_match key_actions
 				{
 					Constraint *n = makeNode(Constraint);
 
@@ -4230,10 +4231,11 @@ ColConstraintElem:
 					n->pktable = $2;
 					n->fk_attrs = NIL;
 					n->pk_attrs = $3;
-					n->fk_matchtype = $4;
-					n->fk_upd_action = ($5)->updateAction->action;
-					n->fk_del_action = ($5)->deleteAction->action;
-					n->fk_del_set_cols = ($5)->deleteAction->cols;
+					n->fk_lockkeyindex = $4;
+					n->fk_matchtype = $5;
+					n->fk_upd_action = ($6)->updateAction->action;
+					n->fk_del_action = ($6)->deleteAction->action;
+					n->fk_del_set_cols = ($6)->deleteAction->cols;
 					n->is_enforced = true;
 					n->skip_validation = false;
 					n->initially_valid = true;
@@ -4491,7 +4493,7 @@ ConstraintElem:
 					$$ = (Node *) n;
 				}
 			| FOREIGN KEY '(' columnList optionalPeriodName ')' REFERENCES qualified_name
-				opt_column_and_period_list key_match key_actions ConstraintAttributeSpec
+				opt_column_and_period_list opt_fklockclause key_match key_actions ConstraintAttributeSpec
 				{
 					Constraint *n = makeNode(Constraint);
 
@@ -4510,11 +4512,12 @@ ConstraintElem:
 						n->pk_attrs = lappend(n->pk_attrs, lsecond($9));
 						n->pk_with_period = true;
 					}
-					n->fk_matchtype = $10;
-					n->fk_upd_action = ($11)->updateAction->action;
-					n->fk_del_action = ($11)->deleteAction->action;
-					n->fk_del_set_cols = ($11)->deleteAction->cols;
-					processCASbits($12, @12, "FOREIGN KEY",
+					n->fk_lockkeyindex = $10;
+					n->fk_matchtype = $11;
+					n->fk_upd_action = ($12)->updateAction->action;
+					n->fk_del_action = ($12)->deleteAction->action;
+					n->fk_del_set_cols = ($12)->deleteAction->cols;
+					processCASbits($13, @13, "FOREIGN KEY",
 								   &n->deferrable, &n->initdeferred,
 								   &n->is_enforced, &n->skip_validation, NULL,
 								   yyscanner);
@@ -4617,6 +4620,14 @@ opt_c_include:	INCLUDE '(' columnList ')'			{ $$ = $3; }
 			 |		/* EMPTY */						{ $$ = NIL; }
 		;
 
+fklockclause:	LOCK_P KEY INDEX					{ $$ = true; }
+			 |	LOCK_P ROWS							{ $$ = false; }
+			 ;
+
+opt_fklockclause:	fklockclause					{ $$ = $1; }
+			 |		/* EMPTY */						{ $$ = false; }
+			 ;
+
 key_match:  MATCH FULL
 			{
 				$$ = FKCONSTR_MATCH_FULL;
diff --git a/src/backend/utils/adt/ri_triggers.c b/src/backend/utils/adt/ri_triggers.c
index dc89c686394..1633714fa9a 100644
--- a/src/backend/utils/adt/ri_triggers.c
+++ b/src/backend/utils/adt/ri_triggers.c
@@ -42,6 +42,7 @@
 #include "miscadmin.h"
 #include "parser/parse_coerce.h"
 #include "parser/parse_relation.h"
+#include "storage/lmgr.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/datum.h"
@@ -56,6 +57,7 @@
 #include "utils/ruleutils.h"
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
+#include "storage/locktag.h"
 
 /*
  * Local definitions
@@ -97,6 +99,9 @@
 #define RI_TRIGTYPE_UPDATE 2
 #define RI_TRIGTYPE_DELETE 3
 
+#define RI_FKEY_INDEXLOCK_KEY_READ		RowExclusiveLock
+#define RI_FKEY_INDEXLOCK_KEY_UPDATE	ShareRowExclusiveLock
+
 typedef struct FastPathMeta FastPathMeta;
 
 /*
@@ -121,6 +126,7 @@ typedef struct RI_ConstraintInfo
 	Oid			fk_relid;		/* referencing relation */
 	char		confupdtype;	/* foreign key's ON UPDATE action */
 	char		confdeltype;	/* foreign key's ON DELETE action */
+	bool		conflockkeyindex;	/* foreign key's marked READ OPTIMIZED */
 	int			ndelsetcols;	/* number of columns referenced in ON DELETE
 								 * SET clause */
 	int16		confdelsetcols[RI_MAX_NUMKEYS]; /* attnums of cols to set on
@@ -449,6 +455,17 @@ RI_FKey_check(TriggerData *trigdata)
 			break;
 	}
 
+	/*
+	 * Lock the backing index if the FK was so defined, to avoid concurrent
+	 * removal of referenced key values without having to lock all contained
+	 * key values' rows.
+	 */
+	if (riinfo->conflockkeyindex)
+	{
+		Assert(OidIsValid(riinfo->conindid));
+		LockRelationOid(riinfo->conindid, RI_FKEY_INDEXLOCK_KEY_READ);
+	}
+
 	/*
 	 * Fast path: probe the PK unique index directly, bypassing SPI.
 	 *
@@ -518,6 +535,10 @@ RI_FKey_check(TriggerData *trigdata)
 		 *  HAVING $n <@ range_agg(x1.r)
 		 * Note if FOR KEY SHARE ever allows GROUP BY and HAVING
 		 * we can make this a bit simpler.
+		 *
+		 * For read-optimized FKs we can skip the FOR KEY SHARE
+		 * row locks, because the keyspace that's referenced is
+		 * locked by the RowExclusive lock on the backing index 
 		 * ----------
 		 */
 		initStringInfo(&querybuf);
@@ -554,7 +575,8 @@ RI_FKey_check(TriggerData *trigdata)
 			querysep = "AND";
 			queryoids[i] = fk_type;
 		}
-		appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
+		if (!riinfo->conflockkeyindex)
+			appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
 		if (riinfo->hasperiod)
 		{
 			Oid			fk_type = RIAttType(fk_rel, riinfo->fk_attnums[riinfo->nkeys - 1]);
@@ -689,6 +711,10 @@ ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
 		 *  HAVING $n <@ range_agg(x1.r)
 		 * Note if FOR KEY SHARE ever allows GROUP BY and HAVING
 		 * we can make this a bit simpler.
+		 *
+		 * For read-optimized FKs we can skip the FOR KEY SHARE
+		 * row locks, because the keyspace that's referenced is
+		 * locked by the RowExclusive lock on the backing index 
 		 * ----------
 		 */
 		initStringInfo(&querybuf);
@@ -723,7 +749,10 @@ ri_Check_Pk_Match(Relation pk_rel, Relation fk_rel,
 			querysep = "AND";
 			queryoids[i] = pk_type;
 		}
-		appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
+
+		if (!riinfo->conflockkeyindex)
+			appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
+
 		if (riinfo->hasperiod)
 		{
 			Oid			fk_type = RIAttType(fk_rel, riinfo->fk_attnums[riinfo->nkeys - 1]);
@@ -862,6 +891,18 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
 	pk_rel = trigdata->tg_relation;
 	oldslot = trigdata->tg_trigslot;
 
+	/*
+	 * Lock the backing index if the FK was so defined, to avoid concurrent
+	 * insertions of referenced key values without other backends having to
+	 * lock all referenced key values' rows.
+	 */
+	if (riinfo->conflockkeyindex)
+	{
+		Assert(OidIsValid(riinfo->conindid));
+
+		LockRelationOid(riinfo->conindid, RI_FKEY_INDEXLOCK_KEY_UPDATE);
+	}
+
 	/*
 	 * If another PK row now exists providing the old key values, we should
 	 * not do anything.  However, this check should only be made in the NO
@@ -952,6 +993,10 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
 		 * We need the coalesce in case the first subquery returns no rows.
 		 * We need the second subquery because FOR KEY SHARE doesn't support
 		 * aggregate queries.
+		 *
+		 * For read-optimized FKs we can skip the FOR KEY SHARE
+		 * row locks, because the keyspace that's referenced is
+		 * locked by the RowExclusive lock on the backing index 
 		 */
 		if (riinfo->hasperiod && is_no_action)
 		{
@@ -1001,7 +1046,9 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
 				querysep = "AND";
 				queryoids[i] = pk_type;
 			}
-			appendStringInfoString(&replacementsbuf, " FOR KEY SHARE OF y) y2)");
+
+			if (!riinfo->conflockkeyindex)
+				appendStringInfoString(&replacementsbuf, " FOR KEY SHARE OF y) y2)");
 
 			ri_GenerateQual(&querybuf, "",
 							intersectbuf.data, fk_period_type,
@@ -1011,7 +1058,8 @@ ri_restrict(TriggerData *trigdata, bool is_no_action)
 			appendStringInfoString(&querybuf, ", false)");
 		}
 
-		appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
+		if (!riinfo->conflockkeyindex)
+			appendStringInfoString(&querybuf, " FOR KEY SHARE OF x");
 
 		/* Prepare and save the plan */
 		qplan = ri_PlanCheck(querybuf.data, riinfo->nkeys, queryoids,
@@ -1069,6 +1117,18 @@ RI_FKey_cascade_del(PG_FUNCTION_ARGS)
 	pk_rel = trigdata->tg_relation;
 	oldslot = trigdata->tg_trigslot;
 
+	/*
+	 * Lock the backing index if the FK was so defined, to avoid concurrent
+	 * insertions of referenced key values without other backends having to
+	 * lock all referenced key values' rows.
+	 */
+	if (riinfo->conflockkeyindex)
+	{
+		Assert(OidIsValid(riinfo->conindid));
+
+		LockRelationOid(riinfo->conindid, RI_FKEY_INDEXLOCK_KEY_UPDATE);
+	}
+
 	SPI_connect();
 
 	/* Fetch or prepare a saved plan for the cascaded delete */
@@ -1174,6 +1234,18 @@ RI_FKey_cascade_upd(PG_FUNCTION_ARGS)
 	newslot = trigdata->tg_newslot;
 	oldslot = trigdata->tg_trigslot;
 
+	/*
+	 * Lock the backing index if the FK was so defined, to avoid concurrent
+	 * insertions of referenced key values without other backends having to
+	 * lock all referenced key values' rows.
+	 */
+	if (riinfo->conflockkeyindex)
+	{
+		Assert(OidIsValid(riinfo->conindid));
+
+		LockRelationOid(riinfo->conindid, RI_FKEY_INDEXLOCK_KEY_UPDATE);
+	}
+
 	SPI_connect();
 
 	/* Fetch or prepare a saved plan for the cascaded update */
@@ -1346,6 +1418,18 @@ ri_set(TriggerData *trigdata, bool is_set_null, int tgkind)
 	pk_rel = trigdata->tg_relation;
 	oldslot = trigdata->tg_trigslot;
 
+	/*
+	 * Lock the backing index if the FK was so defined, to avoid concurrent
+	 * insertions of referenced key values without other backends having to
+	 * lock all referenced key values' rows.
+	 */
+	if (riinfo->conflockkeyindex)
+	{
+		Assert(OidIsValid(riinfo->conindid));
+
+		LockRelationOid(riinfo->conindid, RI_FKEY_INDEXLOCK_KEY_UPDATE);
+	}
+
 	SPI_connect();
 
 	/*
@@ -2449,6 +2533,7 @@ ri_LoadConstraintInfo(Oid constraintOid)
 	riinfo->confupdtype = conForm->confupdtype;
 	riinfo->confdeltype = conForm->confdeltype;
 	riinfo->confmatchtype = conForm->confmatchtype;
+	riinfo->conflockkeyindex = conForm->conflockkeyindex;
 	riinfo->hasperiod = conForm->conperiod;
 
 	DeconstructFkConstraintRow(tup,
@@ -3121,27 +3206,44 @@ ri_FastPathFlushArray(RI_FastPathEntry *fpentry, TupleTableSlot *fk_slot,
 		bool		concurrently_updated;
 		ScanKeyData recheck_skey[1];
 
-		if (!ri_LockPKTuple(pk_rel, pk_slot, snapshot, &concurrently_updated))
-			continue;
-
-		/* Extract the PK value from the matched and locked tuple */
-		found_val = slot_getattr(pk_slot, riinfo->pk_attnums[0], &found_null);
-		Assert(!found_null);
-
-		if (concurrently_updated)
+		/*
+		 * If the FK is configured to lock the index (with conflockkeyindex),
+		 * then we don't need to lock tuples nor check for concurrent
+		 * updates, as those would've had to lock the index to be effective.
+		 * We do need to take a predicate lock on the heap page, though, 
+		 */
+		if (riinfo->conflockkeyindex)
 		{
-			/*
-			 * Build a single-key scankey for recheck.  We need the actual PK
-			 * value that was found, not the FK search value.
-			 */
-			ScanKeyEntryInitialize(&recheck_skey[0], 0, 1,
-								   fpmeta->strats[0],
-								   fpmeta->subtypes[0],
-								   idx_rel->rd_indcollation[0],
-								   fpmeta->regops[0],
-								   found_val);
-			if (!recheck_matched_pk_tuple(idx_rel, recheck_skey, 1, pk_slot))
+			/* nothing to do */
+		}
+		else
+		{
+			if (!ri_LockPKTuple(pk_rel, pk_slot, snapshot,
+								&concurrently_updated))
 				continue;
+
+			/* Extract the PK value from the matched and locked tuple */
+			found_val = slot_getattr(pk_slot, riinfo->pk_attnums[0],
+									 &found_null);
+			Assert(!found_null);
+
+			if (concurrently_updated)
+			{
+				/*
+				 * Build a single-key scankey for recheck.  We need the actual PK
+				 * value that was found, not the FK search value.
+				 */
+				ScanKeyEntryInitialize(&recheck_skey[0], 0, 1,
+									   fpmeta->strats[0],
+									   fpmeta->subtypes[0],
+									   idx_rel->rd_indcollation[0],
+									   fpmeta->regops[0],
+									   found_val);
+
+				if (!recheck_matched_pk_tuple(idx_rel, recheck_skey, 1,
+											  pk_slot))
+					continue;
+			}
 		}
 
 		/*
@@ -3191,8 +3293,16 @@ ri_FastPathProbeOne(Relation pk_rel, Relation idx_rel,
 	{
 		bool		concurrently_updated;
 
-		if (ri_LockPKTuple(pk_rel, slot, snapshot,
-						   &concurrently_updated))
+		if (riinfo->conflockkeyindex)
+		{
+			/*
+			 * index_getscan_next only produces visible tuples; and our
+			 * read-lock on the index precludes any concurrent key removals.
+			 */
+			found = true;
+		}
+		else if (ri_LockPKTuple(pk_rel, slot, snapshot,
+								&concurrently_updated))
 		{
 			if (concurrently_updated)
 				found = recheck_matched_pk_tuple(idx_rel, skey, nkeys, slot);
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index 88de5c0481c..c9c3babf4e4 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/src/backend/utils/adt/ruleutils.c
@@ -2650,6 +2650,9 @@ pg_get_constraintdef_worker(Oid constraintId, bool fullCommand,
 
 				appendStringInfoChar(&buf, ')');
 
+				if (conForm->conflockkeyindex)
+					appendStringInfoString(&buf, " LOCK KEY INDEX");
+
 				/* Add match type */
 				switch (conForm->confmatchtype)
 				{
diff --git a/src/include/catalog/pg_constraint.h b/src/include/catalog/pg_constraint.h
index 1b7fedf1750..f5e4cd7ddcc 100644
--- a/src/include/catalog/pg_constraint.h
+++ b/src/include/catalog/pg_constraint.h
@@ -100,6 +100,7 @@ CATALOG(pg_constraint,2606,ConstraintRelationId)
 	char		confupdtype;	/* foreign key's ON UPDATE action */
 	char		confdeltype;	/* foreign key's ON DELETE action */
 	char		confmatchtype;	/* foreign key's match type */
+	bool		conflockkeyindex; /* foreign key's index-vs-tuple lock option */
 
 	/* Has a local definition (hence, do not drop when coninhcount is 0) */
 	bool		conislocal;
@@ -244,6 +245,7 @@ extern Oid	CreateConstraintEntry(const char *constraintName,
 								  int foreignNKeys,
 								  char foreignUpdateType,
 								  char foreignDeleteType,
+								  bool foreignLockKeyIndex,
 								  const int16 *fkDeleteSetCols,
 								  int numFkDeleteSetCols,
 								  char foreignMatchType,
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 91377a6cde3..43e245ec438 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2978,6 +2978,7 @@ typedef struct Constraint
 	char		fk_matchtype;	/* FULL, PARTIAL, SIMPLE */
 	char		fk_upd_action;	/* ON UPDATE action */
 	char		fk_del_action;	/* ON DELETE action */
+	bool		fk_lockkeyindex;	/* Locks the key index, not individual rows */
 	List	   *fk_del_set_cols;	/* ON DELETE SET NULL/DEFAULT (col1, col2) */
 	List	   *old_conpfeqop;	/* pg_constraint.conpfeqop of my former self */
 	Oid			old_pktable_oid;	/* pg_constraint.confrelid of my former
diff --git a/src/test/isolation/expected/fk-lock-key-index-no-conflicts.out b/src/test/isolation/expected/fk-lock-key-index-no-conflicts.out
new file mode 100644
index 00000000000..1e92c4119f2
--- /dev/null
+++ b/src/test/isolation/expected/fk-lock-key-index-no-conflicts.out
@@ -0,0 +1,131 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s2b s2ukey s1b s1i s2c s1c s2s s1s
+step s2b: BEGIN;
+step s2ukey: UPDATE parent SET parent_key = 3 WHERE parent_key = 2;
+step s1b: BEGIN;
+step s1i: INSERT INTO child VALUES (1, 1); <waiting ...>
+step s2c: COMMIT;
+step s1i: <... completed>
+step s1c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         1|foo
+         3|bar
+(2 rows)
+
+step s1s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+        1|         1
+(1 row)
+
+
+starting permutation: s2b s2uaux s1b s1i s2c s1c s2s s1s
+step s2b: BEGIN;
+step s2uaux: UPDATE parent SET aux = 'baz' WHERE parent_key = 2;
+step s1b: BEGIN;
+step s1i: INSERT INTO child VALUES (1, 1);
+step s2c: COMMIT;
+step s1c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         1|foo
+         2|baz
+(2 rows)
+
+step s1s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+        1|         1
+(1 row)
+
+
+starting permutation: s2b s2ukey s1b s1i s2ukey2 s2c s1c s2s s1s
+step s2b: BEGIN;
+step s2ukey: UPDATE parent SET parent_key = 3 WHERE parent_key = 2;
+step s1b: BEGIN;
+step s1i: INSERT INTO child VALUES (1, 1); <waiting ...>
+step s2ukey2: UPDATE parent SET parent_key = 2 WHERE parent_key = 3;
+step s2c: COMMIT;
+step s1i: <... completed>
+step s1c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         1|foo
+         2|bar
+(2 rows)
+
+step s1s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+        1|         1
+(1 row)
+
+
+starting permutation: s2b s2ukey s3b s3i s2c s3c s2s s3s
+step s2b: BEGIN;
+step s2ukey: UPDATE parent SET parent_key = 3 WHERE parent_key = 2;
+step s3b: BEGIN ISOLATION LEVEL REPEATABLE READ;
+step s3i: INSERT INTO child VALUES (2, 1); <waiting ...>
+step s2c: COMMIT;
+step s3i: <... completed>
+step s3c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         1|foo
+         3|bar
+(2 rows)
+
+step s3s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+        2|         1
+(1 row)
+
+
+starting permutation: s2b s2dkey s3b s3i s2c s3c s2s s3s
+step s2b: BEGIN;
+step s2dkey: DELETE FROM parent WHERE parent_key = 2;
+step s3b: BEGIN ISOLATION LEVEL REPEATABLE READ;
+step s3i: INSERT INTO child VALUES (2, 1); <waiting ...>
+step s2c: COMMIT;
+step s3i: <... completed>
+step s3c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         1|foo
+(1 row)
+
+step s3s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+        2|         1
+(1 row)
+
+
+starting permutation: s2b s2uaux s3b s3i s2c s3c s2s s3s
+step s2b: BEGIN;
+step s2uaux: UPDATE parent SET aux = 'baz' WHERE parent_key = 2;
+step s3b: BEGIN ISOLATION LEVEL REPEATABLE READ;
+step s3i: INSERT INTO child VALUES (2, 1);
+step s2c: COMMIT;
+step s3c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         1|foo
+         2|baz
+(2 rows)
+
+step s3s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+        2|         1
+(1 row)
+
diff --git a/src/test/isolation/expected/fk-lock-key-index-with-conflicts.out b/src/test/isolation/expected/fk-lock-key-index-with-conflicts.out
new file mode 100644
index 00000000000..84937399249
--- /dev/null
+++ b/src/test/isolation/expected/fk-lock-key-index-with-conflicts.out
@@ -0,0 +1,125 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s2b s2ukey s1b s1i s2c s1c s2s s1s
+step s2b: BEGIN;
+step s2ukey: UPDATE parent SET parent_key = 2 WHERE parent_key = 1;
+step s1b: BEGIN;
+step s1i: INSERT INTO child VALUES (1, 1); <waiting ...>
+step s2c: COMMIT;
+step s1i: <... completed>
+ERROR:  insert or update on table "child" violates foreign key constraint "child_parent_key_fkey"
+step s1c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         2|foo
+(1 row)
+
+step s1s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+(0 rows)
+
+
+starting permutation: s2b s2uaux s1b s1i s2c s1c s2s s1s
+step s2b: BEGIN;
+step s2uaux: UPDATE parent SET aux = 'bar' WHERE parent_key = 1;
+step s1b: BEGIN;
+step s1i: INSERT INTO child VALUES (1, 1);
+step s2c: COMMIT;
+step s1c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         1|bar
+(1 row)
+
+step s1s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+        1|         1
+(1 row)
+
+
+starting permutation: s2b s2ukey s1b s1i s2ukey2 s2c s1c s2s s1s
+step s2b: BEGIN;
+step s2ukey: UPDATE parent SET parent_key = 2 WHERE parent_key = 1;
+step s1b: BEGIN;
+step s1i: INSERT INTO child VALUES (1, 1); <waiting ...>
+step s2ukey2: UPDATE parent SET parent_key = 1 WHERE parent_key = 2;
+step s2c: COMMIT;
+step s1i: <... completed>
+step s1c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         1|foo
+(1 row)
+
+step s1s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+        1|         1
+(1 row)
+
+
+starting permutation: s2b s2ukey s3b s3i s2c s3c s2s s3s
+step s2b: BEGIN;
+step s2ukey: UPDATE parent SET parent_key = 2 WHERE parent_key = 1;
+step s3b: BEGIN ISOLATION LEVEL REPEATABLE READ;
+step s3i: INSERT INTO child VALUES (2, 1); <waiting ...>
+step s2c: COMMIT;
+step s3i: <... completed>
+ERROR:  could not serialize access due to concurrent update
+step s3c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         2|foo
+(1 row)
+
+step s3s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+(0 rows)
+
+
+starting permutation: s2b s2dkey s3b s3i s2c s3c s2s s3s
+step s2b: BEGIN;
+step s2dkey: DELETE FROM parent WHERE parent_key = 1;
+step s3b: BEGIN ISOLATION LEVEL REPEATABLE READ;
+step s3i: INSERT INTO child VALUES (2, 1); <waiting ...>
+step s2c: COMMIT;
+step s3i: <... completed>
+ERROR:  could not serialize access due to concurrent delete
+step s3c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+(0 rows)
+
+step s3s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+(0 rows)
+
+
+starting permutation: s2b s2uaux s3b s3i s2c s3c s2s s3s
+step s2b: BEGIN;
+step s2uaux: UPDATE parent SET aux = 'bar' WHERE parent_key = 1;
+step s3b: BEGIN ISOLATION LEVEL REPEATABLE READ;
+step s3i: INSERT INTO child VALUES (2, 1);
+step s2c: COMMIT;
+step s3c: COMMIT;
+step s2s: SELECT * FROM parent;
+parent_key|aux
+----------+---
+         1|bar
+(1 row)
+
+step s3s: SELECT * FROM child;
+child_key|parent_key
+---------+----------
+        2|         1
+(1 row)
+
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index 15c33fad4c5..b9a63a23e2c 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -38,6 +38,8 @@ test: fk-snapshot
 test: fk-snapshot-2
 test: fk-snapshot-3
 test: fk-concurrent-pk-upd
+test: fk-lock-key-index-no-conflicts
+test: fk-lock-key-index-with-conflicts
 test: subxid-overflow
 test: eval-plan-qual
 test: eval-plan-qual-trigger
diff --git a/src/test/isolation/specs/fk-lock-key-index-no-conflicts.spec b/src/test/isolation/specs/fk-lock-key-index-no-conflicts.spec
new file mode 100644
index 00000000000..412d3d98825
--- /dev/null
+++ b/src/test/isolation/specs/fk-lock-key-index-no-conflicts.spec
@@ -0,0 +1,57 @@
+# Tests that an INSERT on referencing table correctly fails when
+# the referenced value disappears due to a concurrent update
+setup
+{
+  CREATE TABLE parent (
+    parent_key int PRIMARY KEY,
+    aux   text NOT NULL
+  );
+
+  CREATE TABLE child (
+    child_key int PRIMARY KEY,
+    parent_key int8 NOT NULL REFERENCES parent LOCK KEY INDEX
+  );
+
+  INSERT INTO parent VALUES (1, 'foo');
+  INSERT INTO parent VALUES (2, 'bar');
+}
+
+teardown
+{
+  DROP TABLE parent, child;
+}
+
+session s1
+step s1b  { BEGIN; }
+step s1i { INSERT INTO child VALUES (1, 1); }
+step s1c { COMMIT; }
+step s1s { SELECT * FROM child; }
+
+session s2
+step s2b  { BEGIN; }
+step s2ukey { UPDATE parent SET parent_key = 3 WHERE parent_key = 2; }
+step s2uaux { UPDATE parent SET aux = 'baz' WHERE parent_key = 2; }
+step s2ukey2 { UPDATE parent SET parent_key = 2 WHERE parent_key = 3; }
+step s2dkey { DELETE FROM parent WHERE parent_key = 2; }
+step s2c { COMMIT; }
+step s2s { SELECT * FROM parent; }
+
+session s3
+step s3b { BEGIN ISOLATION LEVEL REPEATABLE READ; }
+step s3i { INSERT INTO child VALUES (2, 1); }
+step s3c { COMMIT; }
+step s3s { SELECT * FROM child; }
+
+# fail
+permutation s2b s2ukey s1b s1i s2c s1c s2s s1s
+# ok
+permutation s2b s2uaux s1b s1i s2c s1c s2s s1s
+# ok
+permutation s2b s2ukey s1b s1i s2ukey2 s2c s1c s2s s1s
+
+# RR: key update -> serialization failure
+permutation s2b s2ukey s3b s3i s2c s3c s2s s3s
+# RR: key delete -> serialization failure
+permutation s2b s2dkey s3b s3i s2c s3c s2s s3s
+# RR: non-key update -> old version visible via transaction snapshot
+permutation s2b s2uaux s3b s3i s2c s3c s2s s3s
diff --git a/src/test/isolation/specs/fk-lock-key-index-with-conflicts.spec b/src/test/isolation/specs/fk-lock-key-index-with-conflicts.spec
new file mode 100644
index 00000000000..8cf715af19c
--- /dev/null
+++ b/src/test/isolation/specs/fk-lock-key-index-with-conflicts.spec
@@ -0,0 +1,56 @@
+# Tests that an INSERT on referencing table correctly fails when
+# the referenced value disappears due to a concurrent update
+setup
+{
+  CREATE TABLE parent (
+    parent_key int PRIMARY KEY,
+    aux   text NOT NULL
+  );
+
+  CREATE TABLE child (
+    child_key int PRIMARY KEY,
+    parent_key int8 NOT NULL REFERENCES parent LOCK KEY INDEX
+  );
+
+  INSERT INTO parent VALUES (1, 'foo');
+}
+
+teardown
+{
+  DROP TABLE parent, child;
+}
+
+session s1
+step s1b  { BEGIN; }
+step s1i { INSERT INTO child VALUES (1, 1); }
+step s1c { COMMIT; }
+step s1s { SELECT * FROM child; }
+
+session s2
+step s2b  { BEGIN; }
+step s2ukey { UPDATE parent SET parent_key = 2 WHERE parent_key = 1; }
+step s2uaux { UPDATE parent SET aux = 'bar' WHERE parent_key = 1; }
+step s2ukey2 { UPDATE parent SET parent_key = 1 WHERE parent_key = 2; }
+step s2dkey { DELETE FROM parent WHERE parent_key = 1; }
+step s2c { COMMIT; }
+step s2s { SELECT * FROM parent; }
+
+session s3
+step s3b { BEGIN ISOLATION LEVEL REPEATABLE READ; }
+step s3i { INSERT INTO child VALUES (2, 1); }
+step s3c { COMMIT; }
+step s3s { SELECT * FROM child; }
+
+# fail
+permutation s2b s2ukey s1b s1i s2c s1c s2s s1s
+# ok
+permutation s2b s2uaux s1b s1i s2c s1c s2s s1s
+# ok
+permutation s2b s2ukey s1b s1i s2ukey2 s2c s1c s2s s1s
+
+# RR: key update -> serialization failure
+permutation s2b s2ukey s3b s3i s2c s3c s2s s3s
+# RR: key delete -> serialization failure
+permutation s2b s2dkey s3b s3i s2c s3c s2s s3s
+# RR: non-key update -> old version visible via transaction snapshot
+permutation s2b s2uaux s3b s3i s2c s3c s2s s3s
-- 
2.50.1 (Apple Git-155)

