From 031c4afa42b35c23d604062c12717f0be68c2138 Mon Sep 17 00:00:00 2001
From: Mikhail Nikalayeu <mihailnikalayeu@gmail.com>
Date: Thu, 27 Aug 2026 12:40:30 +0200
Subject: [PATCH v1] Don't re-derive in the scan what the apply worker knew
 about its index

FindLogicalRepLocalIndex() picks the index to search the local relation by, either as the relation's replica identity or
primary key, or as one usable for a REPLICA IDENTITY FULL remote relation. Which of the two it is decides whether the
first index match is the tuple or every match has to be compared against the search slot. Only the OID is kept, though,
so RelationFindReplTupleByIndex() works that out a second time on its own, by testing the index it is handed against
GetRelationIdentityOrPK().

The second answer need not match the first. Apply holds only RowExclusiveLock, which conflicts with neither REINDEX
CONCURRENTLY swapping the index for its rebuilt copy nor DROP INDEX CONCURRENTLY marking it invalid, so either can
commit in between ExecOpenIndices() being where the invalidation lands. The scan then compares whole rows against a
search slot that carries one only under REPLICA IDENTITY FULL, nothing matches, and the change is silently dropped as an
update_missing conflict. Assertion builds trip over the identity check just above instead.

Record which of the two it was in the relation map entry, next to the OID, and pass it down, settling the question once.

Oversight in 89e46da5e51.

Author: Mikhail Nikalayeu <mihailnikalayeu@gmail.com>
Reviewed-by: Amit Kapila <amit.kapila16@gmail.com>

Discussion: ................
Backpatch-through: 16, where it was introduced
---
 src/backend/executor/execReplication.c        |  17 +-
 src/backend/replication/logical/relation.c    |  19 +-
 src/backend/replication/logical/worker.c      |  41 +++-
 src/include/executor/executor.h               |   2 +
 src/include/replication/logicalrelation.h     |   5 +
 .../subscription/t/032_subscribe_use_index.pl | 212 ++++++++++++++++++
 6 files changed, 274 insertions(+), 22 deletions(-)

diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index b2ca5cbf117..6910d7b5bb0 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -177,9 +177,17 @@ should_refetch_tuple(TM_Result res, TM_FailureData *tmfd)
  *
  * If a matching tuple is found, lock it with lockmode, fill the slot with its
  * contents, and return true.  Return false otherwise.
+ *
+ * 'isIdxSafeToSkipDuplicates' says how the scan's matches are to be treated:
+ * if true, the index determines the tuple all by itself and the first match
+ * is taken; if false, every match is compared against 'searchslot', which must
+ * then carry a complete row -- the remote relation has REPLICA IDENTITY FULL.
+ * The caller decides; deriving it here would let concurrent DDL flip it after
+ * the index was chosen.
  */
 bool
 RelationFindReplTupleByIndex(Relation rel, Oid idxoid,
+							 bool isIdxSafeToSkipDuplicates,
 							 LockTupleMode lockmode,
 							 TupleTableSlot *searchslot,
 							 TupleTableSlot *outslot)
@@ -192,13 +200,10 @@ RelationFindReplTupleByIndex(Relation rel, Oid idxoid,
 	Relation	idxrel;
 	bool		found;
 	TypeCacheEntry **eq = NULL;
-	bool		isIdxSafeToSkipDuplicates;
 
 	/* Open the index. */
 	idxrel = index_open(idxoid, RowExclusiveLock);
 
-	isIdxSafeToSkipDuplicates = (GetRelationIdentityOrPK(rel) == idxoid);
-
 	InitDirtySnapshot(snap);
 
 	/* Build scan key. */
@@ -629,9 +634,12 @@ RelationFindDeletedTupleInfoSeq(Relation rel, TupleTableSlot *searchslot,
 /*
  * Similar to RelationFindDeletedTupleInfoSeq() but using index scan to locate
  * the deleted tuple.
+ *
+ * 'isIdxSafeToSkipDuplicates' works as in RelationFindReplTupleByIndex().
  */
 bool
 RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid,
+									bool isIdxSafeToSkipDuplicates,
 									TupleTableSlot *searchslot,
 									TransactionId oldestxmin,
 									TransactionId *delete_xid,
@@ -644,7 +652,6 @@ RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid,
 	IndexScanDesc scan;
 	TupleTableSlot *scanslot;
 	TypeCacheEntry **eq = NULL;
-	bool		isIdxSafeToSkipDuplicates;
 	TupleDesc	desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(rel);
 
 	Assert(equalTupleDescs(desc, searchslot->tts_tupleDescriptor));
@@ -654,8 +661,6 @@ RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid,
 	*delete_time = 0;
 	*delete_origin = InvalidReplOriginId;
 
-	isIdxSafeToSkipDuplicates = (GetRelationIdentityOrPK(rel) == idxoid);
-
 	scanslot = table_slot_create(rel, NULL);
 
 	idxrel = index_open(idxoid, RowExclusiveLock);
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index 8ffd2583afb..29e5e46b480 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -56,7 +56,7 @@ typedef struct LogicalRepPartMapEntry
 } LogicalRepPartMapEntry;
 
 static Oid	FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel,
-									 AttrMap *attrMap);
+									 AttrMap *attrMap, bool *isidentity);
 
 /*
  * Relcache invalidation callback for our relation map cache.
@@ -497,7 +497,8 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
 		 * on the relation).
 		 */
 		entry->localindexoid = FindLogicalRepLocalIndex(entry->localrel, remoterel,
-														entry->attrmap);
+														entry->attrmap,
+														&entry->isidentity);
 
 		entry->localrelvalid = true;
 	}
@@ -764,7 +765,8 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root,
 	 * anything in the LogicalRepPartMapContext (hence CacheMemoryContext).
 	 */
 	entry->localindexoid = FindLogicalRepLocalIndex(partrel, remoterel,
-													entry->attrmap);
+													entry->attrmap,
+													&entry->isidentity);
 
 	entry->localrelvalid = true;
 
@@ -925,13 +927,19 @@ GetRelationIdentityOrPK(Relation rel)
 /*
  * Returns the index oid if we can use an index for subscriber. Otherwise,
  * returns InvalidOid.
+ *
+ * '*isidentity' is set to whether the returned index was chosen as the
+ * relation's replica identity or primary key, rather than as one usable
+ * for a REPLICA IDENTITY FULL remote relation.
  */
 static Oid
 FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel,
-						 AttrMap *attrMap)
+						 AttrMap *attrMap, bool *isidentity)
 {
 	Oid			idxoid;
 
+	*isidentity = false;
+
 	/*
 	 * We never need index oid for partitioned tables, always rely on leaf
 	 * partition's index.
@@ -944,7 +952,10 @@ FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel,
 	 */
 	idxoid = GetRelationIdentityOrPK(localrel);
 	if (OidIsValid(idxoid))
+	{
+		*isidentity = true;
 		return idxoid;
+	}
 
 	if (remoterel->replident == REPLICA_IDENTITY_FULL)
 	{
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 0efed3a6f77..448e2717e24 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -290,6 +290,7 @@
 #include "tcop/tcopprot.h"
 #include "utils/acl.h"
 #include "utils/guc.h"
+#include "utils/injection_point.h"
 #include "utils/inval.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
@@ -610,18 +611,22 @@ static void apply_handle_update_internal(ApplyExecutionData *edata,
 										 ResultRelInfo *relinfo,
 										 TupleTableSlot *remoteslot,
 										 LogicalRepTupleData *newtup,
-										 Oid localindexoid);
+										 Oid localindexoid,
+										 bool isidentity);
 static void apply_handle_delete_internal(ApplyExecutionData *edata,
 										 ResultRelInfo *relinfo,
 										 TupleTableSlot *remoteslot,
-										 Oid localindexoid);
+										 Oid localindexoid,
+										 bool isidentity);
 static bool FindReplTupleInLocalRel(ApplyExecutionData *edata, Relation localrel,
 									LogicalRepRelation *remoterel,
 									Oid localidxoid,
+									bool isidentity,
 									TupleTableSlot *remoteslot,
 									TupleTableSlot **localslot);
 static bool FindDeletedTupleInLocalRel(Relation localrel,
 									   Oid localidxoid,
+									   bool isidentity,
 									   TupleTableSlot *remoteslot,
 									   TransactionId *delete_xid,
 									   ReplOriginId *delete_origin,
@@ -2925,7 +2930,8 @@ apply_handle_update(StringInfo s)
 								   remoteslot, &newtup, CMD_UPDATE);
 	else
 		apply_handle_update_internal(edata, edata->targetRelInfo,
-									 remoteslot, &newtup, rel->localindexoid);
+									 remoteslot, &newtup, rel->localindexoid,
+									 rel->isidentity);
 
 	finish_edata(edata);
 
@@ -2950,7 +2956,7 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 							 ResultRelInfo *relinfo,
 							 TupleTableSlot *remoteslot,
 							 LogicalRepTupleData *newtup,
-							 Oid localindexoid)
+							 Oid localindexoid, bool isidentity)
 {
 	EState	   *estate = edata->estate;
 	LogicalRepRelMapEntry *relmapentry = edata->targetRel;
@@ -2962,11 +2968,14 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 	MemoryContext oldctx;
 
 	EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL);
+
+	INJECTION_POINT("apply-update-before-open-indices", NULL);
+
 	ExecOpenIndices(relinfo, false);
 
 	found = FindReplTupleInLocalRel(edata, localrel,
 									&relmapentry->remoterel,
-									localindexoid,
+									localindexoid, isidentity,
 									remoteslot, &localslot);
 
 	/*
@@ -3020,7 +3029,8 @@ apply_handle_update_internal(ApplyExecutionData *edata,
 		 * Detecting whether the tuple was recently deleted or never existed
 		 * is crucial to avoid misleading the user during conflict handling.
 		 */
-		if (FindDeletedTupleInLocalRel(localrel, localindexoid, remoteslot,
+		if (FindDeletedTupleInLocalRel(localrel, localindexoid, isidentity,
+									   remoteslot,
 									   &conflicttuple.xmin,
 									   &conflicttuple.origin,
 									   &conflicttuple.ts) &&
@@ -3122,7 +3132,8 @@ apply_handle_delete(StringInfo s)
 
 		ExecOpenIndices(relinfo, false);
 		apply_handle_delete_internal(edata, relinfo,
-									 remoteslot, rel->localindexoid);
+									 remoteslot, rel->localindexoid,
+									 rel->isidentity);
 		ExecCloseIndices(relinfo);
 	}
 
@@ -3148,7 +3159,7 @@ static void
 apply_handle_delete_internal(ApplyExecutionData *edata,
 							 ResultRelInfo *relinfo,
 							 TupleTableSlot *remoteslot,
-							 Oid localindexoid)
+							 Oid localindexoid, bool isidentity)
 {
 	EState	   *estate = edata->estate;
 	Relation	localrel = relinfo->ri_RelationDesc;
@@ -3166,6 +3177,7 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
 		   RelationGetIndexList(localrel) == NIL);
 
 	found = FindReplTupleInLocalRel(edata, localrel, remoterel, localindexoid,
+									isidentity,
 									remoteslot, &localslot);
 
 	/* If found delete it. */
@@ -3216,6 +3228,7 @@ static bool
 FindReplTupleInLocalRel(ApplyExecutionData *edata, Relation localrel,
 						LogicalRepRelation *remoterel,
 						Oid localidxoid,
+						bool isidentity,
 						TupleTableSlot *remoteslot,
 						TupleTableSlot **localslot)
 {
@@ -3239,14 +3252,14 @@ FindReplTupleInLocalRel(ApplyExecutionData *edata, Relation localrel,
 		Relation	idxrel = index_open(localidxoid, AccessShareLock);
 
 		/* Index must be PK, RI, or usable for REPLICA IDENTITY FULL tables */
-		Assert(GetRelationIdentityOrPK(localrel) == localidxoid ||
+		Assert(isidentity ||
 			   (remoterel->replident == REPLICA_IDENTITY_FULL &&
 				IsIndexUsableForReplicaIdentityFull(idxrel,
 													edata->targetRel->attrmap)));
 		index_close(idxrel, AccessShareLock);
 #endif
 
-		found = RelationFindReplTupleByIndex(localrel, localidxoid,
+		found = RelationFindReplTupleByIndex(localrel, localidxoid, isidentity,
 											 LockTupleExclusive,
 											 remoteslot, *localslot);
 	}
@@ -3308,7 +3321,7 @@ IsIndexUsableForFindingDeletedTuple(Oid localindexoid,
  * '*delete_origin', and '*delete_time' respectively.
  */
 static bool
-FindDeletedTupleInLocalRel(Relation localrel, Oid localidxoid,
+FindDeletedTupleInLocalRel(Relation localrel, Oid localidxoid, bool isidentity,
 						   TupleTableSlot *remoteslot,
 						   TransactionId *delete_xid, ReplOriginId *delete_origin,
 						   TimestampTz *delete_time)
@@ -3377,6 +3390,7 @@ FindDeletedTupleInLocalRel(Relation localrel, Oid localidxoid,
 	if (OidIsValid(localidxoid) &&
 		IsIndexUsableForFindingDeletedTuple(localidxoid, oldestxmin))
 		return RelationFindDeletedTupleInfoByIndex(localrel, localidxoid,
+												   isidentity,
 												   remoteslot, oldestxmin,
 												   delete_xid, delete_origin,
 												   delete_time);
@@ -3479,7 +3493,8 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 		case CMD_DELETE:
 			apply_handle_delete_internal(edata, partrelinfo,
 										 remoteslot_part,
-										 part_entry->localindexoid);
+										 part_entry->localindexoid,
+										 part_entry->isidentity);
 			break;
 
 		case CMD_UPDATE:
@@ -3502,6 +3517,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 				found = FindReplTupleInLocalRel(edata, partrel,
 												&part_entry->remoterel,
 												part_entry->localindexoid,
+												part_entry->isidentity,
 												remoteslot_part, &localslot);
 				if (!found)
 				{
@@ -3515,6 +3531,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
 					 */
 					if (FindDeletedTupleInLocalRel(partrel,
 												   part_entry->localindexoid,
+												   part_entry->isidentity,
 												   remoteslot_part,
 												   &conflicttuple.xmin,
 												   &conflicttuple.origin,
diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h
index 190e8a4897a..f285fca50d1 100644
--- a/src/include/executor/executor.h
+++ b/src/include/executor/executor.h
@@ -775,6 +775,7 @@ extern void check_exclusion_constraint(Relation heap, Relation index,
  * prototypes from functions in execReplication.c
  */
 extern bool RelationFindReplTupleByIndex(Relation rel, Oid idxoid,
+										 bool isIdxSafeToSkipDuplicates,
 										 LockTupleMode lockmode,
 										 TupleTableSlot *searchslot,
 										 TupleTableSlot *outslot);
@@ -787,6 +788,7 @@ extern bool RelationFindDeletedTupleInfoSeq(Relation rel,
 											ReplOriginId *delete_origin,
 											TimestampTz *delete_time);
 extern bool RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid,
+												bool isIdxSafeToSkipDuplicates,
 												TupleTableSlot *searchslot,
 												TransactionId oldestxmin,
 												TransactionId *delete_xid,
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index efe0f9d6031..52f29dda4f6 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -33,6 +33,11 @@ typedef struct LogicalRepRelMapEntry
 	AttrMap    *attrmap;		/* map of local attributes to remote ones */
 	bool		updatable;		/* Can apply updates/deletes? */
 	Oid			localindexoid;	/* which index to use, or InvalidOid if none */
+	bool		isidentity;		/* localindexoid was chosen as the
+								 * relation's replica identity or
+								 * primary key, rather than as one
+								 * usable for a REPLICA IDENTITY
+								 * FULL remote relation */
 
 	/* Sync state. */
 	char		state;
diff --git a/src/test/subscription/t/032_subscribe_use_index.pl b/src/test/subscription/t/032_subscribe_use_index.pl
index 1ccd36ac227..21491c68528 100644
--- a/src/test/subscription/t/032_subscribe_use_index.pl
+++ b/src/test/subscription/t/032_subscribe_use_index.pl
@@ -606,6 +606,218 @@ $node_subscriber->safe_psql('postgres', "DROP TABLE test_replica_id_full");
 # Testcase end: Subscription can use hash index
 # =============================================================================
 
+# =============================================================================
+# Testcase start: Subscription keeps using an index that concurrent DDL has
+# demoted from replica identity
+#
+# REINDEX CONCURRENTLY and DROP INDEX CONCURRENTLY hold locks that do not
+# conflict with the apply worker's, so either can commit after the worker has
+# taken the index out of its relation map entry.  The index goes on finding
+# the row, so the change must still be applied through it, not dropped as a
+# missing-tuple conflict.
+
+SKIP:
+{
+	skip 'Injection points not supported by this build', 8
+	  unless $ENV{enable_injection_points} eq 'yes';
+	skip 'Extension injection_points not installed', 8
+	  unless $node_subscriber->check_extension('injection_points');
+
+	$node_subscriber->safe_psql('postgres',
+		'CREATE EXTENSION injection_points');
+
+	# create tables pub and sub
+	$node_publisher->safe_psql('postgres',
+		"CREATE TABLE test_reindex (x int PRIMARY KEY, y int)");
+	$node_subscriber->safe_psql('postgres',
+		"CREATE TABLE test_reindex (x int PRIMARY KEY, y int)");
+
+	# insert some initial data
+	$node_publisher->safe_psql('postgres',
+		"INSERT INTO test_reindex SELECT i, i FROM generate_series(1,20) i");
+
+	# create pub/sub
+	$node_publisher->safe_psql('postgres',
+		"CREATE PUBLICATION tap_pub_reindex FOR TABLE test_reindex");
+	$node_subscriber->safe_psql('postgres',
+		"CREATE SUBSCRIPTION tap_sub_reindex CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_reindex"
+	);
+
+	# wait for initial table synchronization to finish
+	$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+	my $old_index_oid = $node_subscriber->safe_psql('postgres',
+		"SELECT 'test_reindex_pkey'::regclass::oid");
+
+	# The rebuild goes first and stops at the swap.  The other order does
+	# not work: REINDEX CONCURRENTLY waits for older snapshots, so a worker
+	# paused mid-transaction would block it rather than race it.
+	my $reindex = $node_subscriber->background_psql('postgres');
+	$reindex->query_safe(
+		q[
+		SELECT injection_points_set_local();
+		SELECT injection_points_attach('reindex-relation-concurrently-before-swap', 'wait');
+	]);
+	$reindex->query_until(
+		qr/starting_reindex/, q[
+		\echo starting_reindex
+		REINDEX INDEX CONCURRENTLY test_reindex_pkey;
+	]);
+	$node_subscriber->wait_for_event('client backend',
+		'reindex-relation-concurrently-before-swap');
+
+	# Now let the worker take the index and stop before opening the
+	# relation's indexes.  The point is attached server-wide: the apply
+	# worker is not a session this test can attach anything in.
+	$node_subscriber->safe_psql('postgres',
+		"SELECT injection_points_attach('apply-update-before-open-indices', 'wait')"
+	);
+
+	$node_publisher->safe_psql('postgres',
+		"UPDATE test_reindex SET y = 99 WHERE x = 7");
+
+	$node_subscriber->wait_for_event(
+		'logical replication apply worker',
+		'apply-update-before-open-indices');
+
+	# Let the rebuild swap the index.  Only the swap is needed: dropping the
+	# old index waits for the apply worker's lock anyway.
+	$node_subscriber->safe_psql('postgres',
+		"SELECT injection_points_wakeup('reindex-relation-concurrently-before-swap')"
+	);
+	$node_subscriber->poll_query_until('postgres',
+		"SELECT 'test_reindex_pkey'::regclass::oid <> $old_index_oid")
+	  or die "timed out waiting for the identity index to be swapped";
+
+	# Release the worker.  The index it holds is no longer the identity.
+	$node_subscriber->safe_psql(
+		'postgres',
+		"SELECT injection_points_wakeup('apply-update-before-open-indices');
+		 SELECT injection_points_detach('apply-update-before-open-indices');"
+	);
+	ok($reindex->quit, 'REINDEX CONCURRENTLY completes');
+
+	$node_publisher->wait_for_catchup($appname);
+
+	# The update must have been applied, not dropped as a missing tuple.
+	$result = $node_subscriber->safe_psql('postgres',
+		"SELECT y FROM test_reindex WHERE x = 7");
+	is($result, qq(99), 'update applied through the index in force now');
+
+	# And the worker is still alive to apply the next one.
+	$node_publisher->safe_psql('postgres',
+		"UPDATE test_reindex SET y = 123 WHERE x = 8");
+	$node_publisher->wait_for_catchup($appname);
+	$result = $node_subscriber->safe_psql('postgres',
+		"SELECT y FROM test_reindex WHERE x = 8");
+	is($result, qq(123), 'replication continues');
+
+	# The same window without a rebuild: DROP INDEX CONCURRENTLY clears
+	# indisvalid and indisreplident and commits that before waiting for the
+	# lock apply holds, leaving relreplident set to 'i' with no index
+	# claiming to be that identity.  The drop gets no further while apply
+	# holds the table, so the index is still complete and maintained.
+	$node_publisher->safe_psql(
+		'postgres', q[
+		CREATE TABLE test_dropri (x int NOT NULL, y int);
+		CREATE UNIQUE INDEX test_dropri_ri ON test_dropri (x);
+		ALTER TABLE test_dropri REPLICA IDENTITY USING INDEX test_dropri_ri;
+		INSERT INTO test_dropri SELECT i, i FROM generate_series(1,20) i;
+		CREATE PUBLICATION tap_pub_dropri FOR TABLE test_dropri;
+	]);
+	$node_subscriber->safe_psql(
+		'postgres', q[
+		CREATE TABLE test_dropri (x int NOT NULL, y int);
+		CREATE UNIQUE INDEX test_dropri_ri ON test_dropri (x);
+		ALTER TABLE test_dropri REPLICA IDENTITY USING INDEX test_dropri_ri;
+	]);
+	$node_subscriber->safe_psql('postgres',
+		"CREATE SUBSCRIPTION tap_sub_dropri CONNECTION '$publisher_connstr application_name=dropri' PUBLICATION tap_pub_dropri"
+	);
+	$node_subscriber->wait_for_subscription_sync($node_publisher, 'dropri');
+
+	$node_subscriber->safe_psql('postgres',
+		"SELECT injection_points_attach('apply-update-before-open-indices', 'wait')"
+	);
+	$node_publisher->safe_psql('postgres',
+		"UPDATE test_dropri SET y = 99 WHERE x = 7");
+	$node_subscriber->wait_for_event(
+		'logical replication apply worker',
+		'apply-update-before-open-indices');
+
+	# This commits the loss of the replica identity, then parks waiting for
+	# the apply worker's lock on the table.
+	my $log_offset = -s $node_subscriber->logfile;
+	my $drop = $node_subscriber->background_psql('postgres');
+	$drop->query_until(
+		qr/starting_drop/, q[
+		\echo starting_drop
+		DROP INDEX CONCURRENTLY test_dropri_ri;
+	]);
+	$node_subscriber->poll_query_until('postgres',
+			"SELECT count(*) = 0 FROM pg_index"
+		  . " WHERE indrelid = 'test_dropri'::regclass AND indisreplident")
+	  or die "timed out waiting for the identity index to be invalidated";
+
+	$node_subscriber->safe_psql(
+		'postgres',
+		"SELECT injection_points_detach('apply-update-before-open-indices');
+		 SELECT injection_points_wakeup('apply-update-before-open-indices');"
+	);
+	# The straddling change goes through, found by the demoted index.
+	$node_publisher->wait_for_catchup('dropri');
+	$result = $node_subscriber->safe_psql('postgres',
+		"SELECT y FROM test_dropri WHERE x = 7");
+	is($result, qq(99), 'change straddling the drop is applied');
+	ok($drop->quit, 'DROP INDEX CONCURRENTLY completes');
+
+	# From the next change on the entry is rebuilt, finds no replica
+	# identity, and apply stops with the usual error.
+	$node_publisher->safe_psql('postgres',
+		"UPDATE test_dropri SET y = 123 WHERE x = 8");
+	ok( $node_subscriber->poll_query_until(
+			'postgres', q[
+			SELECT apply_error_count > 0 FROM pg_stat_subscription_stats
+			WHERE subname = 'tap_sub_dropri']),
+		'later changes wait for a replica identity');
+	like(
+		slurp_file($node_subscriber->logfile, $log_offset),
+		qr/logical replication target relation "public\.test_dropri" has neither REPLICA IDENTITY index nor PRIMARY KEY/,
+		'and say why');
+
+	# Give the relation a replica identity again and they resume.
+	$node_subscriber->safe_psql(
+		'postgres', q[
+		CREATE UNIQUE INDEX test_dropri_ri2 ON test_dropri (x);
+		ALTER TABLE test_dropri REPLICA IDENTITY USING INDEX test_dropri_ri2;
+	]);
+	$node_publisher->wait_for_catchup('dropri');
+	$result = $node_subscriber->safe_psql('postgres',
+		"SELECT y FROM test_dropri WHERE x = 8");
+	is($result, qq(123), 'replication resumes');
+
+	$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_dropri");
+	$node_publisher->safe_psql('postgres', "DROP TABLE test_dropri");
+	$node_subscriber->safe_psql('postgres',
+		"DROP SUBSCRIPTION tap_sub_dropri");
+	$node_subscriber->safe_psql('postgres', "DROP TABLE test_dropri");
+
+	# cleanup pub
+	$node_publisher->safe_psql('postgres',
+		"DROP PUBLICATION tap_pub_reindex");
+	$node_publisher->safe_psql('postgres', "DROP TABLE test_reindex");
+	# cleanup sub
+	$node_subscriber->safe_psql('postgres',
+		"DROP SUBSCRIPTION tap_sub_reindex");
+	$node_subscriber->safe_psql('postgres', "DROP TABLE test_reindex");
+	$node_subscriber->safe_psql('postgres',
+		"DROP EXTENSION injection_points");
+}
+
+# Testcase end: Subscription keeps using an index that concurrent DDL has
+# demoted from replica identity
+# =============================================================================
+
 $node_subscriber->stop('fast');
 $node_publisher->stop('fast');
 
-- 
2.43.0

