Fix infer_arbiter_index during concurrent index operations

Previously, we would only consider indexes marked indisvalid as usable
for INSERT ON CONFLICT.  But that's problematic during CREATE INDEX
CONCURRENTLY and REINDEX CONCURRENTLY, because concurrent transactions
would end up with inconsistents lists of inferred indexes, leading to
deadlocks and spurious errors about unique key violations (because two
transactions are operating on different indexes for the speculative
insertion tokens).  Change this function to return indexes even if
invalid.  This fixes the spurious errors and deadlocks.

Because such indexes might not be complete, we still need uniqueness to
be verified in a different way.  We do that by requiring that at least
one index marked valid is part of the set of indexes returned.  It is
that index that is going to help ensure that the inserted tuple is
indeed unique.

This does not fix similar problems occurring with partitioned tables or
with named constraints.  These problems will be fixed in follow-up
commits.

We have no user report of this problem, even though it exists in all
branches.  Because of that and given that the fix is somewhat tricky, I
decided not to backpatch for now.

Author: Mihail Nikalayeu <mihailnikalayeu@gmail.com>
Reviewed-by: Michael Paquier <michael@paquier.xyz>
Reviewed-by: Álvaro Herrera <alvherre@kurilemu.de>
Discussion: https://postgr.es/m/CANtu0ogv+6wqRzPK241jik4U95s1pW3MCZ3rX5ZqbFdUysz7Qw@mail.gmail.com

src/backend/commands/indexcmds.c

@@ -1789,6 +1789,7 @@ DefineIndex(Oid tableId,
 	 * before the reference snap was taken, we have to wait out any
 	 * transactions that might have older snapshots.
 	 */
+	INJECTION_POINT("define-index-before-set-valid", NULL);
 	pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
 								 PROGRESS_CREATEIDX_PHASE_WAIT_3);
 	WaitForOlderSnapshots(limitXmin, true);
@@ -4229,6 +4230,7 @@ ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const Rein
 	 * indexes with the correct names.
 	 */
 
+	INJECTION_POINT("reindex-relation-concurrently-before-swap", NULL);
 	StartTransactionCommand();
 
 	/*
@@ -4307,6 +4309,7 @@ ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const Rein
 	 * index_drop() for more details.
 	 */
 
+	INJECTION_POINT("reindex-relation-concurrently-before-set-dead", NULL);
 	pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
 								 PROGRESS_CREATEIDX_PHASE_WAIT_4);
 	WaitForLockersMultiple(lockTags, AccessExclusiveLock, true);

src/backend/executor/execIndexing.c

@@ -114,6 +114,7 @@
 #include "executor/executor.h"
 #include "nodes/nodeFuncs.h"
 #include "storage/lmgr.h"
+#include "utils/injection_point.h"
 #include "utils/multirangetypes.h"
 #include "utils/rangetypes.h"
 #include "utils/snapmgr.h"
@@ -943,6 +944,13 @@ retry:
 
 	ExecDropSingleTupleTableSlot(existing_slot);
 
+#ifdef USE_INJECTION_POINTS
+	if (conflict)
+		INJECTION_POINT("check-exclusion-or-unique-constraint-conflict", NULL);
+	else
+		INJECTION_POINT("check-exclusion-or-unique-constraint-no-conflict", NULL);
+#endif
+
 	return !conflict;
 }
 

src/backend/executor/execPartition.c

@@ -722,8 +722,9 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
 
 		/*
 		 * If the resulting lists are of inequal length, something is wrong.
-		 * (This shouldn't happen, since arbiter index selection should not
+		 * XXX This may happen because we don't match the lists correctly when
-		 * pick up an invalid index.)
+		 * a partitioned index is being processed by REINDEX CONCURRENTLY.
+		 * FIXME later.
 		 */
 		if (list_length(rootResultRelInfo->ri_onConflictArbiterIndexes) !=
 			list_length(arbiterIndexes))

src/backend/executor/nodeModifyTable.c

@@ -68,6 +68,7 @@
 #include "storage/lmgr.h"
 #include "utils/builtins.h"
 #include "utils/datum.h"
+#include "utils/injection_point.h"
 #include "utils/rel.h"
 #include "utils/snapmgr.h"
 
@@ -1186,6 +1187,7 @@ ExecInsert(ModifyTableContext *context,
 			 * if we're going to go ahead with the insertion, instead of
 			 * waiting for the whole transaction to complete.
 			 */
+			INJECTION_POINT("exec-insert-before-insert-speculative", NULL);
 			specToken = SpeculativeInsertionLockAcquire(GetCurrentTransactionId());
 
 			/* insert the tuple, with the speculative token */

src/backend/optimizer/util/plancat.c

@@ -814,6 +814,7 @@ infer_arbiter_indexes(PlannerInfo *root)
 
 	/* Results */
 	List	   *results = NIL;
+	bool		foundValid = false;
 
 	/*
 	 * Quickly return NIL for ON CONFLICT DO NOTHING without an inference
@@ -907,7 +908,22 @@ infer_arbiter_indexes(PlannerInfo *root)
 		idxRel = index_open(indexoid, rte->rellockmode);
 		idxForm = idxRel->rd_index;
 
-		if (!idxForm->indisvalid)
+		/*
+		 * Ignore indexes that aren't indisready, because we cannot trust
+		 * their catalog structure yet.  However, if any indexes are marked
+		 * indisready but not yet indisvalid, we still consider them, because
+		 * they might turn valid while we're running.  Doing it this way
+		 * allows a concurrent transaction with a slightly later catalog
+		 * snapshot infer the same set of indexes, which is critical to
+		 * prevent spurious 'duplicate key' errors.
+		 *
+		 * However, another critical aspect is that a unique index that isn't
+		 * yet marked indisvalid=true might not be complete yet, meaning it
+		 * wouldn't detect possible duplicate rows.  In order to prevent false
+		 * negatives, we require that we include in the set of inferred
+		 * indexes at least one index that is marked valid.
+		 */
+		if (!idxForm->indisready)
 			goto next;
 
 		/*
@@ -929,10 +945,9 @@ infer_arbiter_indexes(PlannerInfo *root)
 						 errmsg("ON CONFLICT DO UPDATE not supported with exclusion constraints")));
 
 			results = lappend_oid(results, idxForm->indexrelid);
-			list_free(indexList);
+			foundValid |= idxForm->indisvalid;
 			index_close(idxRel, NoLock);
-			table_close(relation, NoLock);
+			break;
-			return results;
 		}
 		else if (indexOidFromConstraint != InvalidOid)
 		{
@@ -1033,6 +1048,7 @@ infer_arbiter_indexes(PlannerInfo *root)
 			goto next;
 
 		results = lappend_oid(results, idxForm->indexrelid);
+		foundValid |= idxForm->indisvalid;
 next:
 		index_close(idxRel, NoLock);
 	}
@@ -1040,7 +1056,8 @@ next:
 	list_free(indexList);
 	table_close(relation, NoLock);
 
-	if (results == NIL)
+	/* We require at least one indisvalid index */
+	if (results == NIL || !foundValid)
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
 				 errmsg("there is no unique or exclusion constraint matching the ON CONFLICT specification")));

src/backend/utils/time/snapmgr.c

@@ -119,6 +119,7 @@
 #include "storage/proc.h"
 #include "storage/procarray.h"
 #include "utils/builtins.h"
+#include "utils/injection_point.h"
 #include "utils/memutils.h"
 #include "utils/resowner.h"
 #include "utils/snapmgr.h"
@@ -458,6 +459,7 @@ InvalidateCatalogSnapshot(void)
 		pairingheap_remove(&RegisteredSnapshots, &CatalogSnapshot->ph_node);
 		CatalogSnapshot = NULL;
 		SnapshotResetXmin();
+		INJECTION_POINT("invalidate-catalog-snapshot-end", NULL);
 	}
 }
 

src/test/modules/injection_points/Makefile

@@ -14,7 +14,12 @@ PGFILEDESC = "injection_points - facility for injection points"
 REGRESS = injection_points hashagg reindex_conc vacuum
 REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress
 
-ISOLATION = basic inplace syscache-update-pruned
+ISOLATION = basic \
+	    inplace \
+	    syscache-update-pruned \
+	    index-concurrently-upsert \
+	    reindex-concurrently-upsert \
+	    index-concurrently-upsert-predicate
 
 TAP_TESTS = 1
 

src/test/modules/injection_points/expected/index-concurrently-upsert-predicate.out

@@ -0,0 +1,86 @@
+Parsed test spec with 4 sessions
+
+starting permutation: s3_start_create_index s1_start_upsert s4_wakeup_define_index_before_set_valid s2_start_upsert s4_wakeup_s1_from_invalidate_catalog_snapshot s4_wakeup_s2 s4_wakeup_s1
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+step s3_start_create_index: 
+	CREATE UNIQUE INDEX CONCURRENTLY tbl_pkey_special_duplicate ON test.tbl(abs(i)) WHERE i < 10000;
+ <waiting ...>
+step s1_start_upsert: 
+	INSERT INTO test.tbl VALUES(13,now()) ON CONFLICT (abs(i)) WHERE i < 100 DO UPDATE SET updated_at = now();
+ <waiting ...>
+step s4_wakeup_define_index_before_set_valid: 
+	SELECT injection_points_detach('define-index-before-set-valid');
+	SELECT injection_points_wakeup('define-index-before-set-valid');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s2_start_upsert: 
+	INSERT INTO test.tbl VALUES(13,now()) ON CONFLICT (abs(i)) WHERE i < 100 DO UPDATE SET updated_at = now();
+ <waiting ...>
+step s4_wakeup_s1_from_invalidate_catalog_snapshot: 
+	SELECT injection_points_detach('invalidate-catalog-snapshot-end');
+	SELECT injection_points_wakeup('invalidate-catalog-snapshot-end');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s4_wakeup_s2: 
+	SELECT injection_points_detach('exec-insert-before-insert-speculative');
+	SELECT injection_points_wakeup('exec-insert-before-insert-speculative');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s4_wakeup_s1: 
+	SELECT injection_points_detach('check-exclusion-or-unique-constraint-no-conflict');
+	SELECT injection_points_wakeup('check-exclusion-or-unique-constraint-no-conflict');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s2_start_upsert: <... completed>
+step s3_start_create_index: <... completed>

src/test/modules/injection_points/expected/index-concurrently-upsert.out

@@ -0,0 +1,86 @@
+Parsed test spec with 4 sessions
+
+starting permutation: s3_start_create_index s1_start_upsert s4_wakeup_define_index_before_set_valid s2_start_upsert s4_wakeup_s1_from_invalidate_catalog_snapshot s4_wakeup_s2 s4_wakeup_s1
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+step s3_start_create_index: 
+	CREATE UNIQUE INDEX CONCURRENTLY tbl_pkey_duplicate ON test.tbl(i);
+ <waiting ...>
+step s1_start_upsert: 
+    INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+ <waiting ...>
+step s4_wakeup_define_index_before_set_valid: 
+	SELECT injection_points_detach('define-index-before-set-valid');
+	SELECT injection_points_wakeup('define-index-before-set-valid');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s2_start_upsert: 
+	INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+ <waiting ...>
+step s4_wakeup_s1_from_invalidate_catalog_snapshot: 
+	SELECT injection_points_detach('invalidate-catalog-snapshot-end');
+	SELECT injection_points_wakeup('invalidate-catalog-snapshot-end');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s4_wakeup_s2: 
+	SELECT injection_points_detach('exec-insert-before-insert-speculative');
+	SELECT injection_points_wakeup('exec-insert-before-insert-speculative');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s4_wakeup_s1: 
+	SELECT injection_points_detach('check-exclusion-or-unique-constraint-no-conflict');
+	SELECT injection_points_wakeup('check-exclusion-or-unique-constraint-no-conflict');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s2_start_upsert: <... completed>
+step s3_start_create_index: <... completed>

src/test/modules/injection_points/expected/reindex-concurrently-upsert.out

@@ -0,0 +1,238 @@
+Parsed test spec with 4 sessions
+
+starting permutation: s3_setup_wait_before_set_dead s3_start_reindex s1_start_upsert s4_wakeup_to_set_dead s2_start_upsert s4_wakeup_s1 s4_wakeup_s2
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+injection_points_set_local
+--------------------------
+                          
+(1 row)
+
+step s3_setup_wait_before_set_dead: 
+	SELECT injection_points_attach('reindex-relation-concurrently-before-set-dead', 'wait');
+
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+step s3_start_reindex: 
+	REINDEX INDEX CONCURRENTLY test.tbl_pkey;
+ <waiting ...>
+step s1_start_upsert: 
+	INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+ <waiting ...>
+step s4_wakeup_to_set_dead: 
+	SELECT injection_points_detach('reindex-relation-concurrently-before-set-dead');
+	SELECT injection_points_wakeup('reindex-relation-concurrently-before-set-dead');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s2_start_upsert: 
+	INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+ <waiting ...>
+step s4_wakeup_s1: 
+	SELECT injection_points_detach('check-exclusion-or-unique-constraint-no-conflict');
+	SELECT injection_points_wakeup('check-exclusion-or-unique-constraint-no-conflict');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_s2: 
+	SELECT injection_points_detach('exec-insert-before-insert-speculative');
+	SELECT injection_points_wakeup('exec-insert-before-insert-speculative');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s2_start_upsert: <... completed>
+step s3_start_reindex: <... completed>
+
+starting permutation: s3_setup_wait_before_swap s3_start_reindex s1_start_upsert s4_wakeup_to_swap s2_start_upsert s4_wakeup_s2 s4_wakeup_s1
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+injection_points_set_local
+--------------------------
+                          
+(1 row)
+
+step s3_setup_wait_before_swap: 
+	SELECT injection_points_attach('reindex-relation-concurrently-before-swap', 'wait');
+
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+step s3_start_reindex: 
+	REINDEX INDEX CONCURRENTLY test.tbl_pkey;
+ <waiting ...>
+step s1_start_upsert: 
+	INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+ <waiting ...>
+step s4_wakeup_to_swap: 
+	SELECT injection_points_detach('reindex-relation-concurrently-before-swap');
+	SELECT injection_points_wakeup('reindex-relation-concurrently-before-swap');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s2_start_upsert: 
+	INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+ <waiting ...>
+step s4_wakeup_s2: 
+	SELECT injection_points_detach('exec-insert-before-insert-speculative');
+	SELECT injection_points_wakeup('exec-insert-before-insert-speculative');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s4_wakeup_s1: 
+	SELECT injection_points_detach('check-exclusion-or-unique-constraint-no-conflict');
+	SELECT injection_points_wakeup('check-exclusion-or-unique-constraint-no-conflict');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s2_start_upsert: <... completed>
+step s3_start_reindex: <... completed>
+
+starting permutation: s3_setup_wait_before_set_dead s3_start_reindex s1_start_upsert s2_start_upsert s4_wakeup_s1 s4_wakeup_to_set_dead s4_wakeup_s2
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+injection_points_set_local
+--------------------------
+                          
+(1 row)
+
+step s3_setup_wait_before_set_dead: 
+	SELECT injection_points_attach('reindex-relation-concurrently-before-set-dead', 'wait');
+
+injection_points_attach
+-----------------------
+                       
+(1 row)
+
+step s3_start_reindex: 
+	REINDEX INDEX CONCURRENTLY test.tbl_pkey;
+ <waiting ...>
+step s1_start_upsert: 
+	INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+ <waiting ...>
+step s2_start_upsert: 
+	INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+ <waiting ...>
+step s4_wakeup_s1: 
+	SELECT injection_points_detach('check-exclusion-or-unique-constraint-no-conflict');
+	SELECT injection_points_wakeup('check-exclusion-or-unique-constraint-no-conflict');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_to_set_dead: 
+	SELECT injection_points_detach('reindex-relation-concurrently-before-set-dead');
+	SELECT injection_points_wakeup('reindex-relation-concurrently-before-set-dead');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s4_wakeup_s2: 
+	SELECT injection_points_detach('exec-insert-before-insert-speculative');
+	SELECT injection_points_wakeup('exec-insert-before-insert-speculative');
+
+injection_points_detach
+-----------------------
+                       
+(1 row)
+
+injection_points_wakeup
+-----------------------
+                       
+(1 row)
+
+step s2_start_upsert: <... completed>
+step s3_start_reindex: <... completed>

src/test/modules/injection_points/meson.build

@@ -48,8 +48,13 @@ tests += {
       'basic',
       'inplace',
       'syscache-update-pruned',
+      'index-concurrently-upsert',
+      'reindex-concurrently-upsert',
+      'index-concurrently-upsert-predicate',
     ],
     'runningcheck': false, # see syscache-update-pruned
+    # Some tests wait for all snapshots, so avoid parallel execution
+    'runningcheck-parallel': false,
   },
   'tap': {
     'env': {
@@ -58,5 +63,7 @@ tests += {
     'tests': [
       't/001_stats.pl',
     ],
+    # The injection points are cluster-wide, so disable installcheck
+    'runningcheck': false,
   },
 }

src/test/modules/injection_points/specs/index-concurrently-upsert-predicate.spec

@@ -0,0 +1,88 @@
+# This test verifies INSERT ON CONFLICT DO UPDATE behavior concurrent with
+# CREATE INDEX CONCURRENTLY a partial index.
+#
+# - s1: UPSERT a tuple
+# - s2: UPSERT the same tuple
+# - s3: CREATE UNIQUE INDEX CONCURRENTLY (with a predicate)
+#
+# - s4: control concurrency via injection points
+
+setup
+{
+	CREATE EXTENSION injection_points;
+	CREATE SCHEMA test;
+	CREATE UNLOGGED TABLE test.tbl(i int, updated_at timestamp);
+	CREATE UNIQUE INDEX tbl_pkey_special ON test.tbl(abs(i)) WHERE i < 1000;
+	ALTER TABLE test.tbl SET (parallel_workers=0);
+}
+
+teardown
+{
+	DROP SCHEMA test CASCADE;
+	DROP EXTENSION injection_points;
+}
+
+session s1
+setup
+{
+	SELECT injection_points_set_local();
+	SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
+	SELECT injection_points_attach('invalidate-catalog-snapshot-end', 'wait');
+}
+step s1_start_upsert
+{
+	INSERT INTO test.tbl VALUES(13,now()) ON CONFLICT (abs(i)) WHERE i < 100 DO UPDATE SET updated_at = now();
+}
+
+session s2
+setup
+{
+	SELECT injection_points_set_local();
+	SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
+}
+step s2_start_upsert
+{
+	INSERT INTO test.tbl VALUES(13,now()) ON CONFLICT (abs(i)) WHERE i < 100 DO UPDATE SET updated_at = now();
+}
+
+session s3
+setup
+{
+	SELECT injection_points_set_local();
+	SELECT injection_points_attach('define-index-before-set-valid', 'wait');
+}
+step s3_start_create_index
+{
+	CREATE UNIQUE INDEX CONCURRENTLY tbl_pkey_special_duplicate ON test.tbl(abs(i)) WHERE i < 10000;
+}
+
+session s4
+step s4_wakeup_s1
+{
+	SELECT injection_points_detach('check-exclusion-or-unique-constraint-no-conflict');
+	SELECT injection_points_wakeup('check-exclusion-or-unique-constraint-no-conflict');
+}
+step s4_wakeup_s1_from_invalidate_catalog_snapshot
+{
+	SELECT injection_points_detach('invalidate-catalog-snapshot-end');
+	SELECT injection_points_wakeup('invalidate-catalog-snapshot-end');
+}
+step s4_wakeup_s2
+{
+	SELECT injection_points_detach('exec-insert-before-insert-speculative');
+	SELECT injection_points_wakeup('exec-insert-before-insert-speculative');
+}
+step s4_wakeup_define_index_before_set_valid
+{
+	SELECT injection_points_detach('define-index-before-set-valid');
+	SELECT injection_points_wakeup('define-index-before-set-valid');
+}
+
+permutation
+	s3_start_create_index(s1_start_upsert, s2_start_upsert)
+	s1_start_upsert
+	s4_wakeup_define_index_before_set_valid
+	s2_start_upsert(s1_start_upsert)
+	s4_wakeup_s1_from_invalidate_catalog_snapshot
+	s4_wakeup_s2
+	s4_wakeup_s1

src/test/modules/injection_points/specs/index-concurrently-upsert.spec

@@ -0,0 +1,87 @@
+# This test verifies INSERT ON CONFLICT DO UPDATE behavior concurrent with
+# CREATE INDEX CONCURRENTLY.
+#
+# - s1: UPSERT a tuple
+# - s2: UPSERT the same tuple
+# - s3: CREATE UNIQUE INDEX CONCURRENTLY
+#
+# - s4: Control concurrency using injection points
+
+setup
+{
+	CREATE EXTENSION injection_points;
+	CREATE SCHEMA test;
+	CREATE UNLOGGED TABLE test.tbl(i int primary key, updated_at timestamp);
+	ALTER TABLE test.tbl SET (parallel_workers=0);
+}
+
+teardown
+{
+	DROP SCHEMA test CASCADE;
+	DROP EXTENSION injection_points;
+}
+
+session s1
+setup
+{
+	SELECT injection_points_set_local();
+	SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
+	SELECT injection_points_attach('invalidate-catalog-snapshot-end', 'wait');
+}
+step s1_start_upsert
+{
+    INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+}
+
+session s2
+setup
+{
+	SELECT injection_points_set_local();
+	SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
+}
+step s2_start_upsert
+{
+	INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+}
+
+session s3
+setup
+{
+	SELECT injection_points_set_local();
+	SELECT injection_points_attach('define-index-before-set-valid', 'wait');
+}
+step s3_start_create_index
+{
+	CREATE UNIQUE INDEX CONCURRENTLY tbl_pkey_duplicate ON test.tbl(i);
+}
+
+session s4
+step s4_wakeup_s1
+{
+	SELECT injection_points_detach('check-exclusion-or-unique-constraint-no-conflict');
+	SELECT injection_points_wakeup('check-exclusion-or-unique-constraint-no-conflict');
+}
+step s4_wakeup_s1_from_invalidate_catalog_snapshot
+{
+	SELECT injection_points_detach('invalidate-catalog-snapshot-end');
+	SELECT injection_points_wakeup('invalidate-catalog-snapshot-end');
+}
+step s4_wakeup_s2
+{
+	SELECT injection_points_detach('exec-insert-before-insert-speculative');
+	SELECT injection_points_wakeup('exec-insert-before-insert-speculative');
+}
+step s4_wakeup_define_index_before_set_valid
+{
+	SELECT injection_points_detach('define-index-before-set-valid');
+	SELECT injection_points_wakeup('define-index-before-set-valid');
+}
+
+permutation
+	s3_start_create_index(s1_start_upsert, s2_start_upsert)
+	s1_start_upsert
+	s4_wakeup_define_index_before_set_valid
+	s2_start_upsert(s1_start_upsert)
+	s4_wakeup_s1_from_invalidate_catalog_snapshot
+	s4_wakeup_s2
+	s4_wakeup_s1

src/test/modules/injection_points/specs/reindex-concurrently-upsert.spec

@@ -0,0 +1,111 @@
+# This test verifies INSERT ON CONFLICT DO UPDATE behavior concurrent with
+# REINDEX CONCURRENTLY.
+#
+# - s1: UPSERT a tuple
+# - s2: UPSERT the same tuple
+# - s3: REINDEX concurrent primary key index
+#
+# - s4: controls concurrency via injection points
+
+setup
+{
+	CREATE EXTENSION injection_points;
+	CREATE SCHEMA test;
+	CREATE UNLOGGED TABLE test.tbl (i int PRIMARY KEY, updated_at timestamp);
+	ALTER TABLE test.tbl SET (parallel_workers=0);
+}
+
+teardown
+{
+	DROP SCHEMA test CASCADE;
+	DROP EXTENSION injection_points;
+}
+
+session s1
+setup
+{
+	SELECT injection_points_set_local();
+	SELECT injection_points_attach('check-exclusion-or-unique-constraint-no-conflict', 'wait');
+}
+step s1_start_upsert
+{
+	INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+}
+
+session s2
+setup
+{
+	SELECT injection_points_set_local();
+	SELECT injection_points_attach('exec-insert-before-insert-speculative', 'wait');
+}
+step s2_start_upsert
+{
+	INSERT INTO test.tbl VALUES (13,now()) ON CONFLICT (i) DO UPDATE SET updated_at = now();
+}
+
+session s3
+setup
+{
+	SELECT injection_points_set_local();
+}
+step s3_setup_wait_before_set_dead
+{
+	SELECT injection_points_attach('reindex-relation-concurrently-before-set-dead', 'wait');
+}
+step s3_setup_wait_before_swap
+{
+	SELECT injection_points_attach('reindex-relation-concurrently-before-swap', 'wait');
+}
+step s3_start_reindex
+{
+	REINDEX INDEX CONCURRENTLY test.tbl_pkey;
+}
+
+session s4
+step s4_wakeup_to_swap
+{
+	SELECT injection_points_detach('reindex-relation-concurrently-before-swap');
+	SELECT injection_points_wakeup('reindex-relation-concurrently-before-swap');
+}
+step s4_wakeup_s1
+{
+	SELECT injection_points_detach('check-exclusion-or-unique-constraint-no-conflict');
+	SELECT injection_points_wakeup('check-exclusion-or-unique-constraint-no-conflict');
+}
+step s4_wakeup_s2
+{
+	SELECT injection_points_detach('exec-insert-before-insert-speculative');
+	SELECT injection_points_wakeup('exec-insert-before-insert-speculative');
+}
+step s4_wakeup_to_set_dead
+{
+	SELECT injection_points_detach('reindex-relation-concurrently-before-set-dead');
+	SELECT injection_points_wakeup('reindex-relation-concurrently-before-set-dead');
+}
+
+permutation
+	s3_setup_wait_before_set_dead
+	s3_start_reindex(s1_start_upsert, s2_start_upsert)
+	s1_start_upsert
+	s4_wakeup_to_set_dead
+	s2_start_upsert(s1_start_upsert)
+	s4_wakeup_s1
+	s4_wakeup_s2
+
+permutation
+	s3_setup_wait_before_swap
+	s3_start_reindex(s1_start_upsert, s2_start_upsert)
+	s1_start_upsert
+	s4_wakeup_to_swap
+	s2_start_upsert(s1_start_upsert)
+	s4_wakeup_s2
+	s4_wakeup_s1
+
+permutation
+	s3_setup_wait_before_set_dead
+	s3_start_reindex(s1_start_upsert, s2_start_upsert)
+	s1_start_upsert
+	s2_start_upsert(s1_start_upsert)
+	s4_wakeup_s1
+	s4_wakeup_to_set_dead
+	s4_wakeup_s2