Hello, everyone! This patch set addresses the issues discussed in this thread.
The main idea behind this fix is that it is safe to consider indisready indexes alongside indisvalid indexes as arbiter indexes. However, it's crucial that at least one fully valid index is present. Why is it necessary to consider indisready during the planning phase? The reason is that these indexes are required for correct processing during the execution phase. If "ready" indexes are skipped as arbiters by one transaction, they may already have become "valid" for another concurrent transaction during its planning phase. As a result, both transactions could concurrently process the UPSERT command with different sets of arbiters (while using the same set of indexes for tuple insertion later). This can lead to unexpected "duplicate key value violates unique constraint" errors and deadlocks. Is it safe to use a "ready" but not yet "valid" index as an arbiter? Yes, as long as at least one "valid" index is also used as an arbiter. The valid index ensures the correctness of the UPSERT logic, while the "ready" index contains an equal or lesser number of tuples, making it safe for speculative insertion. In any case, the insert to that index will be processed during ExecInsertIndexTuples one way or another (with applyNoDupErr or without). Fix is divided into a few patches, each following this logic: 1) The first patch provides specs (and injection points) for the various scenarios related to the issue. 2) The second patch introduces a straightforward change—adding indisready indexes to arbiters alongside indisvalid. However, at least one indisvalid is still required. This resolves simple cases involving REINDEX CONCURRENTLY and CREATE INDEX CONCURRENTLY. 3) The third patch deals with named constraints. Instead of relying solely on the index with the specified name, we attempt to find other indexes that are equivalent in terms of being used as an arbiter. 4) This patch fixes a scenario involving partitioned tables. Special checks are required for partitioned indexes, which may be processed by REINDEX CONCURRENTLY. Additionally, a patch with three extra TAP specifications for stress testing is attached. This patch is not intended for commitment, so I renamed the extension to prevent accidental application in some CI/DI jobs. > Also, it is possible to look at the patches on GitHub: https://github.com/postgres/postgres/compare/master...michail-nikolaev:postgres:reindex_concurrently_with_upsert Best regards, Mikhail.
From 9a4d01018e21d491c967b3378e61abdb06f52c74 Mon Sep 17 00:00:00 2001
From: nkey <n...@toloka.ai>
Date: Sat, 24 Aug 2024 14:04:02 +0200
Subject: [PATCH v3 2/4] Modify the infer_arbiter_indexes function to consider
both indisvalid and indisready indexes. Ensure that at least one indisvalid
index is still required.
The change ensures that all concurrent transactions utilize the same set of indexes as arbiters. This uniformity is required to avoid conditions that could lead to "duplicate key value violates unique constraint" errors during UPSERT operations.
The patch resolves the issues in the following specs:
* reindex_concurrently_upsert
* index_concurrently_upsert
* index_concurrently_upsert_predicate
Despite the patch, the following specs are still affected:
* reindex_concurrently_upsert_partitioned
* reindex_concurrently_upsert_on_constraint
---
src/backend/optimizer/util/plancat.c | 18 +++++++++++++-----
1 file changed, 13 insertions(+), 5 deletions(-)
diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c
index 78a3cfafde..5ffc815ae4 100644
--- a/src/backend/optimizer/util/plancat.c
+++ b/src/backend/optimizer/util/plancat.c
@@ -720,6 +720,7 @@ infer_arbiter_indexes(PlannerInfo *root)
/* Results */
List *results = NIL;
+ bool foundValid = false;
/*
* Quickly return NIL for ON CONFLICT DO NOTHING without an inference
@@ -813,7 +814,13 @@ infer_arbiter_indexes(PlannerInfo *root)
idxRel = index_open(indexoid, rte->rellockmode);
idxForm = idxRel->rd_index;
- if (!idxForm->indisvalid)
+ /*
+ * We need to consider both indisvalid and indisready indexes because
+ * them may become indisvalid before execution phase. It is required
+ * to keep set of indexes used as arbiter to be the same for all
+ * concurrent transactions.
+ */
+ if (!idxForm->indisready)
goto next;
/*
@@ -835,10 +842,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);
- return results;
+ break;
}
else if (indexOidFromConstraint != InvalidOid)
{
@@ -932,6 +938,7 @@ infer_arbiter_indexes(PlannerInfo *root)
goto next;
results = lappend_oid(results, idxForm->indexrelid);
+ foundValid |= idxForm->indisvalid;
next:
index_close(idxRel, NoLock);
}
@@ -939,7 +946,8 @@ next:
list_free(indexList);
table_close(relation, NoLock);
- if (results == NIL)
+ /* It is required to have at least one indisvalid index during the planning. */
+ if (results == NIL || !foundValid)
ereport(ERROR,
(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
errmsg("there is no unique or exclusion constraint matching the ON CONFLICT specification")));
--
2.34.1
From c2566ce93a8e24410ae89c1838a57e310e33cf88 Mon Sep 17 00:00:00 2001
From: nkey <n...@toloka.ai>
Date: Sat, 24 Aug 2024 13:44:32 +0200
Subject: [PATCH v3 1/4] Specs top reproduce the issues with CREATE INDEX
CONCURRENTLY and REINDEX CONCURRENTLY in scenarios involving INSERT ON
CONFLICT DO UPDATE. These tests reproduce different error cases related to
"duplicate key value violates unique constraint" where this error should not
occur by design.
* REINDEX CONCURRENTLY and UPSERT with inferred index
* CREATE INDEX CONCURRENTLY and UPSERT with inferred indexes
* REINDEX CONCURRENTLY on partitioned table
* REINDEX CONCURRENTLY with specified constraint name
* CREATE INDEX CONCURRENTLY with predicates
In each of these scenarios, the expected behavior is that the INSERT ON CONFLICT DO UPDATE should handle conflicts gracefully without raising a "duplicate key value violates unique constraint" error. However, due to the concurrent operations on the indexes, this error is encountered.
---
src/backend/commands/indexcmds.c | 5 +-
src/backend/executor/execIndexing.c | 3 +
src/backend/executor/nodeModifyTable.c | 2 +
src/backend/utils/time/snapmgr.c | 2 +
src/test/modules/injection_points/Makefile | 7 +-
.../expected/index_concurrently_upsert.out | 80 ++++++
.../index_concurrently_upsert_predicate.out | 80 ++++++
.../expected/reindex_concurrently_upsert.out | 238 ++++++++++++++++++
...ndex_concurrently_upsert_on_constraint.out | 238 ++++++++++++++++++
...eindex_concurrently_upsert_partitioned.out | 238 ++++++++++++++++++
src/test/modules/injection_points/meson.build | 5 +
.../specs/index_concurrently_upsert.spec | 68 +++++
.../index_concurrently_upsert_predicate.spec | 70 ++++++
.../specs/reindex_concurrently_upsert.spec | 86 +++++++
...dex_concurrently_upsert_on_constraint.spec | 86 +++++++
...index_concurrently_upsert_partitioned.spec | 88 +++++++
16 files changed, 1294 insertions(+), 2 deletions(-)
create mode 100644 src/test/modules/injection_points/expected/index_concurrently_upsert.out
create mode 100644 src/test/modules/injection_points/expected/index_concurrently_upsert_predicate.out
create mode 100644 src/test/modules/injection_points/expected/reindex_concurrently_upsert.out
create mode 100644 src/test/modules/injection_points/expected/reindex_concurrently_upsert_on_constraint.out
create mode 100644 src/test/modules/injection_points/expected/reindex_concurrently_upsert_partitioned.out
create mode 100644 src/test/modules/injection_points/specs/index_concurrently_upsert.spec
create mode 100644 src/test/modules/injection_points/specs/index_concurrently_upsert_predicate.spec
create mode 100644 src/test/modules/injection_points/specs/reindex_concurrently_upsert.spec
create mode 100644 src/test/modules/injection_points/specs/reindex_concurrently_upsert_on_constraint.spec
create mode 100644 src/test/modules/injection_points/specs/reindex_concurrently_upsert_partitioned.spec
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
index c5a56c75f6..ed7ed0f640 100644
--- a/src/backend/commands/indexcmds.c
+++ b/src/backend/commands/indexcmds.c
@@ -69,6 +69,7 @@
#include "utils/regproc.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
+#include "utils/injection_point.h"
/* non-export function prototypes */
@@ -1776,6 +1777,7 @@ DefineIndex(Oid tableId,
pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
PROGRESS_CREATEIDX_PHASE_WAIT_3);
WaitForOlderSnapshots(limitXmin, true);
+ INJECTION_POINT("define_index_before_set_valid");
/*
* Index can now be marked valid -- update its pg_index entry
@@ -4082,7 +4084,7 @@ ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const Rein
* the same time to make sure we only get constraint violations from the
* indexes with the correct names.
*/
-
+ INJECTION_POINT("reindex_relation_concurrently_before_swap");
StartTransactionCommand();
/*
@@ -4156,6 +4158,7 @@ ReindexRelationConcurrently(const ReindexStmt *stmt, Oid relationOid, const Rein
pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
PROGRESS_CREATEIDX_PHASE_WAIT_4);
WaitForLockersMultiple(lockTags, AccessExclusiveLock, true);
+ INJECTION_POINT("reindex_relation_concurrently_before_set_dead");
foreach(lc, indexIds)
{
diff --git a/src/backend/executor/execIndexing.c b/src/backend/executor/execIndexing.c
index 403a3f4055..5f0957e53f 100644
--- a/src/backend/executor/execIndexing.c
+++ b/src/backend/executor/execIndexing.c
@@ -115,6 +115,7 @@
#include "nodes/nodeFuncs.h"
#include "storage/lmgr.h"
#include "utils/snapmgr.h"
+#include "utils/injection_point.h"
/* waitMode argument to check_exclusion_or_unique_constraint() */
typedef enum
@@ -906,6 +907,8 @@ retry:
econtext->ecxt_scantuple = save_scantuple;
ExecDropSingleTupleTableSlot(existing_slot);
+ if (!conflict)
+ INJECTION_POINT("check_exclusion_or_unique_constraint_no_conflict");
return !conflict;
}
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index 8bf4c80d4a..cdbf0b4d4f 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -69,6 +69,7 @@
#include "utils/datum.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
+#include "utils/injection_point.h"
typedef struct MTTargetRelLookup
@@ -1087,6 +1088,7 @@ ExecInsert(ModifyTableContext *context,
return NULL;
}
}
+ INJECTION_POINT("exec_insert_before_insert_speculative");
/*
* Before we start insertion proper, acquire our "speculative
diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c
index 7d2b34d4f2..3a7357a050 100644
--- a/src/backend/utils/time/snapmgr.c
+++ b/src/backend/utils/time/snapmgr.c
@@ -64,6 +64,7 @@
#include "utils/resowner.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
+#include "utils/injection_point.h"
/*
@@ -426,6 +427,7 @@ InvalidateCatalogSnapshot(void)
pairingheap_remove(&RegisteredSnapshots, &CatalogSnapshot->ph_node);
CatalogSnapshot = NULL;
SnapshotResetXmin();
+ INJECTION_POINT("invalidate_catalog_snapshot_end");
}
}
diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile
index ed28cd13a8..0d5005be1f 100644
--- a/src/test/modules/injection_points/Makefile
+++ b/src/test/modules/injection_points/Makefile
@@ -13,7 +13,12 @@ PGFILEDESC = "injection_points - facility for injection points"
REGRESS = injection_points
REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress
-ISOLATION = inplace
+ISOLATION = inplace \
+ reindex_concurrently_upsert \
+ index_concurrently_upsert \
+ reindex_concurrently_upsert_partitioned \
+ reindex_concurrently_upsert_on_constraint \
+ index_concurrently_upsert_predicate
TAP_TESTS = 1
diff --git a/src/test/modules/injection_points/expected/index_concurrently_upsert.out b/src/test/modules/injection_points/expected/index_concurrently_upsert.out
new file mode 100644
index 0000000000..f39a6d452a
--- /dev/null
+++ b/src/test/modules/injection_points/expected/index_concurrently_upsert.out
@@ -0,0 +1,80 @@
+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_wakeup('define_index_before_set_valid');
+ SELECT injection_points_detach('define_index_before_set_valid');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_create_index: <... completed>
+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_wakeup('invalidate_catalog_snapshot_end');
+ SELECT injection_points_detach('invalidate_catalog_snapshot_end');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s2_start_upsert: <... completed>
+step s4_wakeup_s1:
+ SELECT injection_points_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
diff --git a/src/test/modules/injection_points/expected/index_concurrently_upsert_predicate.out b/src/test/modules/injection_points/expected/index_concurrently_upsert_predicate.out
new file mode 100644
index 0000000000..014d94d7ec
--- /dev/null
+++ b/src/test/modules/injection_points/expected/index_concurrently_upsert_predicate.out
@@ -0,0 +1,80 @@
+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_wakeup('define_index_before_set_valid');
+ SELECT injection_points_detach('define_index_before_set_valid');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_create_index: <... completed>
+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_wakeup('invalidate_catalog_snapshot_end');
+ SELECT injection_points_detach('invalidate_catalog_snapshot_end');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s2_start_upsert: <... completed>
+step s4_wakeup_s1:
+ SELECT injection_points_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
diff --git a/src/test/modules/injection_points/expected/reindex_concurrently_upsert.out b/src/test/modules/injection_points/expected/reindex_concurrently_upsert.out
new file mode 100644
index 0000000000..b7639ff7e6
--- /dev/null
+++ b/src/test/modules/injection_points/expected/reindex_concurrently_upsert.out
@@ -0,0 +1,238 @@
+Parsed test spec with 4 sessions
+
+starting permutation: s3_start_reindex s1_start_upsert s4_wakeup_to_swap s2_start_upsert s4_wakeup_s1 s4_wakeup_s2 s4_wakeup_to_set_dead
+injection_points_attach
+-----------------------
+
+(1 row)
+
+injection_points_attach
+-----------------------
+
+(1 row)
+
+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_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(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_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s2_start_upsert: <... completed>
+step s4_wakeup_to_set_dead:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: <... completed>
+
+starting permutation: s3_start_reindex s2_start_upsert s4_wakeup_to_swap s1_start_upsert s4_wakeup_s1 s4_wakeup_s2 s4_wakeup_to_set_dead
+injection_points_attach
+-----------------------
+
+(1 row)
+
+injection_points_attach
+-----------------------
+
+(1 row)
+
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: REINDEX INDEX CONCURRENTLY test.tbl_pkey; <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_to_swap:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_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_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s2_start_upsert: <... completed>
+step s4_wakeup_to_set_dead:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: <... completed>
+
+starting permutation: s3_start_reindex s4_wakeup_to_swap 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_attach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: REINDEX INDEX CONCURRENTLY test.tbl_pkey; <waiting ...>
+step s4_wakeup_to_swap:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+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_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_to_set_dead:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: <... completed>
+step s2_start_upsert: <... completed>
diff --git a/src/test/modules/injection_points/expected/reindex_concurrently_upsert_on_constraint.out b/src/test/modules/injection_points/expected/reindex_concurrently_upsert_on_constraint.out
new file mode 100644
index 0000000000..dbbb9691cc
--- /dev/null
+++ b/src/test/modules/injection_points/expected/reindex_concurrently_upsert_on_constraint.out
@@ -0,0 +1,238 @@
+Parsed test spec with 4 sessions
+
+starting permutation: s3_start_reindex s1_start_upsert s4_wakeup_to_swap s2_start_upsert s4_wakeup_s1 s4_wakeup_s2 s4_wakeup_to_set_dead
+injection_points_attach
+-----------------------
+
+(1 row)
+
+injection_points_attach
+-----------------------
+
+(1 row)
+
+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 on constraint tbl_pkey do update set updated_at = now(); <waiting ...>
+step s4_wakeup_to_swap:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s2_start_upsert: INSERT INTO test.tbl VALUES(13,now()) on conflict on constraint tbl_pkey do update set updated_at = now(); <waiting ...>
+step s4_wakeup_s1:
+ SELECT injection_points_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s2_start_upsert: <... completed>
+step s4_wakeup_to_set_dead:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: <... completed>
+
+starting permutation: s3_start_reindex s2_start_upsert s4_wakeup_to_swap s1_start_upsert s4_wakeup_s1 s4_wakeup_s2 s4_wakeup_to_set_dead
+injection_points_attach
+-----------------------
+
+(1 row)
+
+injection_points_attach
+-----------------------
+
+(1 row)
+
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: REINDEX INDEX CONCURRENTLY test.tbl_pkey; <waiting ...>
+step s2_start_upsert: INSERT INTO test.tbl VALUES(13,now()) on conflict on constraint tbl_pkey do update set updated_at = now(); <waiting ...>
+step s4_wakeup_to_swap:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: INSERT INTO test.tbl VALUES(13,now()) on conflict on constraint tbl_pkey do update set updated_at = now(); <waiting ...>
+step s4_wakeup_s1:
+ SELECT injection_points_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s2_start_upsert: <... completed>
+step s4_wakeup_to_set_dead:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: <... completed>
+
+starting permutation: s3_start_reindex s4_wakeup_to_swap 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_attach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: REINDEX INDEX CONCURRENTLY test.tbl_pkey; <waiting ...>
+step s4_wakeup_to_swap:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: INSERT INTO test.tbl VALUES(13,now()) on conflict on constraint tbl_pkey do update set updated_at = now(); <waiting ...>
+step s2_start_upsert: INSERT INTO test.tbl VALUES(13,now()) on conflict on constraint tbl_pkey do update set updated_at = now(); <waiting ...>
+step s4_wakeup_s1:
+ SELECT injection_points_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_to_set_dead:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: <... completed>
+step s2_start_upsert: <... completed>
diff --git a/src/test/modules/injection_points/expected/reindex_concurrently_upsert_partitioned.out b/src/test/modules/injection_points/expected/reindex_concurrently_upsert_partitioned.out
new file mode 100644
index 0000000000..b4cedf820c
--- /dev/null
+++ b/src/test/modules/injection_points/expected/reindex_concurrently_upsert_partitioned.out
@@ -0,0 +1,238 @@
+Parsed test spec with 4 sessions
+
+starting permutation: s3_start_reindex s1_start_upsert s4_wakeup_to_swap s2_start_upsert s4_wakeup_s1 s4_wakeup_s2 s4_wakeup_to_set_dead
+injection_points_attach
+-----------------------
+
+(1 row)
+
+injection_points_attach
+-----------------------
+
+(1 row)
+
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: REINDEX INDEX CONCURRENTLY test.tbl_partition_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_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(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_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s2_start_upsert: <... completed>
+step s4_wakeup_to_set_dead:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: <... completed>
+
+starting permutation: s3_start_reindex s2_start_upsert s4_wakeup_to_swap s1_start_upsert s4_wakeup_s1 s4_wakeup_s2 s4_wakeup_to_set_dead
+injection_points_attach
+-----------------------
+
+(1 row)
+
+injection_points_attach
+-----------------------
+
+(1 row)
+
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey; <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_to_swap:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_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_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s2_start_upsert: <... completed>
+step s4_wakeup_to_set_dead:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: <... completed>
+
+starting permutation: s3_start_reindex s4_wakeup_to_swap 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_attach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey; <waiting ...>
+step s4_wakeup_to_swap:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+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_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s1_start_upsert: <... completed>
+step s4_wakeup_to_set_dead:
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s4_wakeup_s2:
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+step s3_start_reindex: <... completed>
+step s2_start_upsert: <... completed>
diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build
index c9e357f644..6391417f4b 100644
--- a/src/test/modules/injection_points/meson.build
+++ b/src/test/modules/injection_points/meson.build
@@ -42,6 +42,11 @@ tests += {
'isolation': {
'specs': [
'inplace',
+ 'reindex_concurrently_upsert',
+ 'index_concurrently_upsert',
+ 'reindex_concurrently_upsert_partitioned',
+ 'reindex_concurrently_upsert_on_constraint',
+ 'index_concurrently_upsert_predicate',
],
},
'tap': {
diff --git a/src/test/modules/injection_points/specs/index_concurrently_upsert.spec b/src/test/modules/injection_points/specs/index_concurrently_upsert.spec
new file mode 100644
index 0000000000..5d6aba9073
--- /dev/null
+++ b/src/test/modules/injection_points/specs/index_concurrently_upsert.spec
@@ -0,0 +1,68 @@
+# Test race conditions involving:
+# - s1: UPSERT a tuple
+# - s2: UPSERT the same tuple
+# - s3: CREATE UNIQUE INDEX CONCURRENTLY
+# - s4: operations with 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_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+}
+step s4_wakeup_s1_from_invalidate_catalog_snapshot {
+ SELECT injection_points_wakeup('invalidate_catalog_snapshot_end');
+ SELECT injection_points_detach('invalidate_catalog_snapshot_end');
+}
+step s4_wakeup_s2 {
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+}
+step s4_wakeup_define_index_before_set_valid {
+ SELECT injection_points_wakeup('define_index_before_set_valid');
+ SELECT injection_points_detach('define_index_before_set_valid');
+}
+
+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
\ No newline at end of file
diff --git a/src/test/modules/injection_points/specs/index_concurrently_upsert_predicate.spec b/src/test/modules/injection_points/specs/index_concurrently_upsert_predicate.spec
new file mode 100644
index 0000000000..2bc3cfd775
--- /dev/null
+++ b/src/test/modules/injection_points/specs/index_concurrently_upsert_predicate.spec
@@ -0,0 +1,70 @@
+# Test race conditions involving:
+# - s1: UPSERT a tuple
+# - s2: UPSERT the same tuple
+# - s3: CREATE UNIQUE INDEX CONCURRENTLY
+# - s4: operations with 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_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+}
+step s4_wakeup_s1_from_invalidate_catalog_snapshot {
+ SELECT injection_points_wakeup('invalidate_catalog_snapshot_end');
+ SELECT injection_points_detach('invalidate_catalog_snapshot_end');
+}
+step s4_wakeup_s2 {
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+}
+step s4_wakeup_define_index_before_set_valid {
+ SELECT injection_points_wakeup('define_index_before_set_valid');
+ SELECT injection_points_detach('define_index_before_set_valid');
+}
+
+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
\ No newline at end of file
diff --git a/src/test/modules/injection_points/specs/reindex_concurrently_upsert.spec b/src/test/modules/injection_points/specs/reindex_concurrently_upsert.spec
new file mode 100644
index 0000000000..c6ad2c4198
--- /dev/null
+++ b/src/test/modules/injection_points/specs/reindex_concurrently_upsert.spec
@@ -0,0 +1,86 @@
+# Test race conditions involving:
+# - s1: UPSERT a tuple
+# - s2: UPSERT the same tuple
+# - s3: REINDEX concurrent primary key index
+# - s4: operations with 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();
+ SELECT injection_points_attach('reindex_relation_concurrently_before_set_dead', 'wait');
+ 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_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+}
+step s4_wakeup_s1 {
+ SELECT injection_points_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+}
+step s4_wakeup_s2 {
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+}
+step s4_wakeup_to_set_dead {
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+}
+
+permutation
+ s3_start_reindex
+ s1_start_upsert
+ s4_wakeup_to_swap
+ s2_start_upsert
+ s4_wakeup_s1
+ s4_wakeup_s2
+ s4_wakeup_to_set_dead
+
+permutation
+ s3_start_reindex
+ s2_start_upsert
+ s4_wakeup_to_swap
+ s1_start_upsert
+ s4_wakeup_s1
+ s4_wakeup_s2
+ s4_wakeup_to_set_dead
+
+permutation
+ s3_start_reindex
+ s4_wakeup_to_swap
+ s1_start_upsert
+ s2_start_upsert
+ s4_wakeup_s1
+ s4_wakeup_to_set_dead
+ s4_wakeup_s2
\ No newline at end of file
diff --git a/src/test/modules/injection_points/specs/reindex_concurrently_upsert_on_constraint.spec b/src/test/modules/injection_points/specs/reindex_concurrently_upsert_on_constraint.spec
new file mode 100644
index 0000000000..fb030f8575
--- /dev/null
+++ b/src/test/modules/injection_points/specs/reindex_concurrently_upsert_on_constraint.spec
@@ -0,0 +1,86 @@
+# Test race conditions involving:
+# - s1: UPSERT a tuple
+# - s2: UPSERT the same tuple
+# - s3: REINDEX concurrent primary key index
+# - s4: operations with 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 on constraint tbl_pkey 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 on constraint tbl_pkey do update set updated_at = now(); }
+
+session s3
+setup {
+ SELECT injection_points_set_local();
+ SELECT injection_points_attach('reindex_relation_concurrently_before_set_dead', 'wait');
+ 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_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+}
+step s4_wakeup_s1 {
+ SELECT injection_points_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+}
+step s4_wakeup_s2 {
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+}
+step s4_wakeup_to_set_dead {
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+}
+
+permutation
+ s3_start_reindex
+ s1_start_upsert
+ s4_wakeup_to_swap
+ s2_start_upsert
+ s4_wakeup_s1
+ s4_wakeup_s2
+ s4_wakeup_to_set_dead
+
+permutation
+ s3_start_reindex
+ s2_start_upsert
+ s4_wakeup_to_swap
+ s1_start_upsert
+ s4_wakeup_s1
+ s4_wakeup_s2
+ s4_wakeup_to_set_dead
+
+permutation
+ s3_start_reindex
+ s4_wakeup_to_swap
+ s1_start_upsert
+ s2_start_upsert
+ s4_wakeup_s1
+ s4_wakeup_to_set_dead
+ s4_wakeup_s2
\ No newline at end of file
diff --git a/src/test/modules/injection_points/specs/reindex_concurrently_upsert_partitioned.spec b/src/test/modules/injection_points/specs/reindex_concurrently_upsert_partitioned.spec
new file mode 100644
index 0000000000..efa55809ae
--- /dev/null
+++ b/src/test/modules/injection_points/specs/reindex_concurrently_upsert_partitioned.spec
@@ -0,0 +1,88 @@
+# Test race conditions involving:
+# - s1: UPSERT a tuple
+# - s2: UPSERT the same tuple
+# - s3: REINDEX concurrent primary key index
+# - s4: operations with injection points
+
+setup
+{
+ CREATE EXTENSION injection_points;
+ CREATE SCHEMA test;
+ CREATE TABLE test.tbl(i int primary key, updated_at timestamp) PARTITION BY RANGE (i);
+ CREATE TABLE test.tbl_partition PARTITION OF test.tbl
+ FOR VALUES FROM (0) TO (10000)
+ WITH (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();
+ SELECT injection_points_attach('reindex_relation_concurrently_before_set_dead', 'wait');
+ SELECT injection_points_attach('reindex_relation_concurrently_before_swap', 'wait');
+}
+step s3_start_reindex { REINDEX INDEX CONCURRENTLY test.tbl_partition_pkey; }
+
+session s4
+step s4_wakeup_to_swap {
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_swap');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_swap');
+}
+step s4_wakeup_s1 {
+ SELECT injection_points_wakeup('check_exclusion_or_unique_constraint_no_conflict');
+ SELECT injection_points_detach('check_exclusion_or_unique_constraint_no_conflict');
+}
+step s4_wakeup_s2 {
+ SELECT injection_points_wakeup('exec_insert_before_insert_speculative');
+ SELECT injection_points_detach('exec_insert_before_insert_speculative');
+}
+step s4_wakeup_to_set_dead {
+ SELECT injection_points_wakeup('reindex_relation_concurrently_before_set_dead');
+ SELECT injection_points_detach('reindex_relation_concurrently_before_set_dead');
+}
+
+permutation
+ s3_start_reindex
+ s1_start_upsert
+ s4_wakeup_to_swap
+ s2_start_upsert
+ s4_wakeup_s1
+ s4_wakeup_s2
+ s4_wakeup_to_set_dead
+
+permutation
+ s3_start_reindex
+ s2_start_upsert
+ s4_wakeup_to_swap
+ s1_start_upsert
+ s4_wakeup_s1
+ s4_wakeup_s2
+ s4_wakeup_to_set_dead
+
+permutation
+ s3_start_reindex
+ s4_wakeup_to_swap
+ s1_start_upsert
+ s2_start_upsert
+ s4_wakeup_s1
+ s4_wakeup_to_set_dead
+ s4_wakeup_s2
\ No newline at end of file
--
2.34.1
From fd2dad875688f2d8f465c4a689940edb1824c39d Mon Sep 17 00:00:00 2001
From: nkey <n...@toloka.ai>
Date: Sat, 24 Aug 2024 14:33:47 +0200
Subject: [PATCH v3 3/4] Modify the infer_arbiter_indexes function to also look
for indexes that match the specified named constraint to be used as arbiters.
This ensures that the same set of arbiter indexes is used for all concurrent
transactions in cases where REINDEX CONCURRENTLY processes an index used as a
named constraint.
The patch resolves the issues in the following specs:
* reindex_concurrently_upsert_on_constraint
Despite the patch, the following specs are still affected:
* reindex_concurrently_upsert_partitioned
---
src/backend/optimizer/util/plancat.c | 121 +++++++++++++++++++--------
1 file changed, 88 insertions(+), 33 deletions(-)
diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c
index 5ffc815ae4..6712efe4dd 100644
--- a/src/backend/optimizer/util/plancat.c
+++ b/src/backend/optimizer/util/plancat.c
@@ -714,9 +714,10 @@ infer_arbiter_indexes(PlannerInfo *root)
List *indexList;
ListCell *l;
- /* Normalized inference attributes and inference expressions: */
- Bitmapset *inferAttrs = NULL;
- List *inferElems = NIL;
+ /* Normalized required attributes and expressions: */
+ Bitmapset *requiredArbiterAttrs = NULL;
+ List *requiredArbiterElems = NIL;
+ List *requiredIndexPredExprs = (List *) onconflict->arbiterWhere;
/* Results */
List *results = NIL;
@@ -755,8 +756,8 @@ infer_arbiter_indexes(PlannerInfo *root)
if (!IsA(elem->expr, Var))
{
- /* If not a plain Var, just shove it in inferElems for now */
- inferElems = lappend(inferElems, elem->expr);
+ /* If not a plain Var, just shove it in requiredArbiterElems for now */
+ requiredArbiterElems = lappend(requiredArbiterElems, elem->expr);
continue;
}
@@ -768,30 +769,76 @@ infer_arbiter_indexes(PlannerInfo *root)
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("whole row unique index inference specifications are not supported")));
- inferAttrs = bms_add_member(inferAttrs,
+ requiredArbiterAttrs = bms_add_member(requiredArbiterAttrs,
attno - FirstLowInvalidHeapAttributeNumber);
}
+ indexList = RelationGetIndexList(relation);
+
/*
* Lookup named constraint's index. This is not immediately returned
- * because some additional sanity checks are required.
+ * because some additional sanity checks are required. Additionally, we
+ * need to process other indexes as potential arbiters to account for
+ * cases where REINDEX CONCURRENTLY is processing an index used as a
+ * named constraint.
*/
if (onconflict->constraint != InvalidOid)
{
indexOidFromConstraint = get_constraint_index(onconflict->constraint);
if (indexOidFromConstraint == InvalidOid)
+ {
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
- errmsg("constraint in ON CONFLICT clause has no associated index")));
+ errmsg("constraint in ON CONFLICT clause has no associated index")));
+ }
+
+ /*
+ * Find the named constraint index to extract its attributes and predicates.
+ * We open all indexes in the loop to avoid deadlock of changed order of locks.
+ * */
+ foreach(l, indexList)
+ {
+ Oid indexoid = lfirst_oid(l);
+ Relation idxRel;
+ Form_pg_index idxForm;
+ AttrNumber natt;
+
+ idxRel = index_open(indexoid, rte->rellockmode);
+ idxForm = idxRel->rd_index;
+
+ if (idxForm->indisready)
+ {
+ if (indexOidFromConstraint == idxForm->indexrelid)
+ {
+ /*
+ * Prepare requirements for other indexes to be used as arbiter together
+ * with indexOidFromConstraint. It is required to involve both equals indexes
+ * in case of REINDEX CONCURRENTLY.
+ */
+ for (natt = 0; natt < idxForm->indnkeyatts; natt++)
+ {
+ int attno = idxRel->rd_index->indkey.values[natt];
+
+ if (attno != 0)
+ requiredArbiterAttrs = bms_add_member(requiredArbiterAttrs,
+ attno - FirstLowInvalidHeapAttributeNumber);
+ }
+ requiredArbiterElems = RelationGetIndexExpressions(idxRel);
+ requiredIndexPredExprs = RelationGetIndexPredicate(idxRel);
+ /* We are done, so, quite the loop. */
+ index_close(idxRel, NoLock);
+ break;
+ }
+ }
+ index_close(idxRel, NoLock);
+ }
}
/*
* Using that representation, iterate through the list of indexes on the
* target relation to try and find a match
*/
- indexList = RelationGetIndexList(relation);
-
foreach(l, indexList)
{
Oid indexoid = lfirst_oid(l);
@@ -840,26 +887,23 @@ infer_arbiter_indexes(PlannerInfo *root)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("ON CONFLICT DO UPDATE not supported with exclusion constraints")));
-
- results = lappend_oid(results, idxForm->indexrelid);
- foundValid |= idxForm->indisvalid;
- index_close(idxRel, NoLock);
- break;
+ goto found;
}
else if (indexOidFromConstraint != InvalidOid)
{
- /* No point in further work for index in named constraint case */
- goto next;
+ /* In the case of "ON constraint_name DO UPDATE" we need to skip non-unique candidates. */
+ if (!idxForm->indisunique && onconflict->action == ONCONFLICT_UPDATE)
+ goto next;
+ } else {
+ /*
+ * Only considering conventional inference at this point (not named
+ * constraints), so index under consideration can be immediately
+ * skipped if it's not unique
+ */
+ if (!idxForm->indisunique)
+ goto next;
}
- /*
- * Only considering conventional inference at this point (not named
- * constraints), so index under consideration can be immediately
- * skipped if it's not unique
- */
- if (!idxForm->indisunique)
- goto next;
-
/* Build BMS representation of plain (non expression) index attrs */
indexedAttrs = NULL;
for (natt = 0; natt < idxForm->indnkeyatts; natt++)
@@ -872,7 +916,7 @@ infer_arbiter_indexes(PlannerInfo *root)
}
/* Non-expression attributes (if any) must match */
- if (!bms_equal(indexedAttrs, inferAttrs))
+ if (!bms_equal(indexedAttrs, requiredArbiterAttrs))
goto next;
/* Expression attributes (if any) must match */
@@ -880,6 +924,10 @@ infer_arbiter_indexes(PlannerInfo *root)
if (idxExprs && varno != 1)
ChangeVarNodes((Node *) idxExprs, 1, varno, 0);
+ /*
+ * If arbiterElems are present, check them. If name >constraint is
+ * present arbiterElems == NIL.
+ */
foreach(el, onconflict->arbiterElems)
{
InferenceElem *elem = (InferenceElem *) lfirst(el);
@@ -917,26 +965,33 @@ infer_arbiter_indexes(PlannerInfo *root)
}
/*
- * Now that all inference elements were matched, ensure that the
+ * In case of the conventional inference involved ensure that the
* expression elements from inference clause are not missing any
* cataloged expressions. This does the right thing when unique
* indexes redundantly repeat the same attribute, or if attributes
* redundantly appear multiple times within an inference clause.
+ *
+ * In the case of named constraint ensure candidate has equal set
+ * of expressions as the named constraint index.
*/
- if (list_difference(idxExprs, inferElems) != NIL)
+ if (list_difference(idxExprs, requiredArbiterElems) != NIL)
goto next;
- /*
- * If it's a partial index, its predicate must be implied by the ON
- * CONFLICT's WHERE clause.
- */
predExprs = RelationGetIndexPredicate(idxRel);
if (predExprs && varno != 1)
ChangeVarNodes((Node *) predExprs, 1, varno, 0);
- if (!predicate_implied_by(predExprs, (List *) onconflict->arbiterWhere, false))
+ /*
+ * If it's a partial index and conventional inference, its predicate must be implied
+ * by the ON CONFLICT's WHERE clause.
+ */
+ if (indexOidFromConstraint == InvalidOid && !predicate_implied_by(predExprs, requiredIndexPredExprs, false))
+ goto next;
+ /* If it's a partial index and named constraint predicates must be equal. */
+ if (indexOidFromConstraint != InvalidOid && list_difference(predExprs, requiredIndexPredExprs) != NIL)
goto next;
+found:
results = lappend_oid(results, idxForm->indexrelid);
foundValid |= idxForm->indisvalid;
next:
--
2.34.1
From 0c51f52341205b52fefa2ce0e7f3513835abf49e Mon Sep 17 00:00:00 2001
From: nkey <n...@toloka.ai>
Date: Sat, 24 Aug 2024 15:19:53 +0200
Subject: [PATCH v3 4/4] Modify the ExecInitPartitionInfo function to consider
partitioned indexes that are potentially processed by REINDEX CONCURRENTLY as
arbiters as well.
This is necessary to ensure that all concurrent transactions use the same set of arbiter indexes.
The patch resolves the issues in the following specs:
* reindex_concurrently_upsert_partitioned
---
src/backend/executor/execPartition.c | 119 ++++++++++++++++++++++++---
1 file changed, 107 insertions(+), 12 deletions(-)
diff --git a/src/backend/executor/execPartition.c b/src/backend/executor/execPartition.c
index 7651886229..aeeee41d5f 100644
--- a/src/backend/executor/execPartition.c
+++ b/src/backend/executor/execPartition.c
@@ -483,6 +483,48 @@ ExecFindPartition(ModifyTableState *mtstate,
return rri;
}
+/*
+ * IsIndexCompatibleAsArbiter
+ * Checks if the indexes are identical in terms of being used
+ * as arbiters for the INSERT ON CONFLICT operation by comparing
+ * them to the provided arbiter index.
+ *
+ * Returns the true if indexes are compatible.
+ */
+static bool
+IsIndexCompatibleAsArbiter(Relation arbiterIndexRelation,
+ IndexInfo *arbiterIndexInfo,
+ Relation indexRelation,
+ IndexInfo *indexInfo)
+{
+ int i;
+
+ if (arbiterIndexInfo->ii_Unique != indexInfo->ii_Unique)
+ return false;
+ /* it is not supported for cases of exclusion constraints. */
+ if (arbiterIndexInfo->ii_ExclusionOps != NULL || indexInfo->ii_ExclusionOps != NULL)
+ return false;
+ if (arbiterIndexRelation->rd_index->indnkeyatts != indexRelation->rd_index->indnkeyatts)
+ return false;
+
+ for (i = 0; i < indexRelation->rd_index->indnkeyatts; i++)
+ {
+ int arbiterAttoNo = arbiterIndexRelation->rd_index->indkey.values[i];
+ int attoNo = indexRelation->rd_index->indkey.values[i];
+ if (arbiterAttoNo != attoNo)
+ return false;
+ }
+
+ if (list_difference(RelationGetIndexExpressions(arbiterIndexRelation),
+ RelationGetIndexExpressions(indexRelation)) != NIL)
+ return false;
+
+ if (list_difference(RelationGetIndexPredicate(arbiterIndexRelation),
+ RelationGetIndexPredicate(indexRelation)) != NIL)
+ return false;
+ return true;
+}
+
/*
* ExecInitPartitionInfo
* Lock the partition and initialize ResultRelInfo. Also setup other
@@ -693,6 +735,8 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
if (rootResultRelInfo->ri_onConflictArbiterIndexes != NIL)
{
List *childIdxs;
+ List *nonAncestorIdxs = NIL;
+ int i, j, additional_arbiters = 0;
childIdxs = RelationGetIndexList(leaf_part_rri->ri_RelationDesc);
@@ -703,23 +747,74 @@ ExecInitPartitionInfo(ModifyTableState *mtstate, EState *estate,
ListCell *lc2;
ancestors = get_partition_ancestors(childIdx);
- foreach(lc2, rootResultRelInfo->ri_onConflictArbiterIndexes)
+ if (ancestors)
{
- if (list_member_oid(ancestors, lfirst_oid(lc2)))
- arbiterIndexes = lappend_oid(arbiterIndexes, childIdx);
+ foreach(lc2, rootResultRelInfo->ri_onConflictArbiterIndexes)
+ {
+ if (list_member_oid(ancestors, lfirst_oid(lc2)))
+ arbiterIndexes = lappend_oid(arbiterIndexes, childIdx);
+ }
}
+ else /* No ancestor was found for that index. Save it for rechecking later. */
+ nonAncestorIdxs = lappend_oid(nonAncestorIdxs, childIdx);
list_free(ancestors);
}
+
+ /*
+ * If any non-ancestor indexes are found, we need to compare them with other
+ * indexes of the relation that will be used as arbiters. This is necessary
+ * when a partitioned index is processed by REINDEX CONCURRENTLY. Both indexes
+ * must be considered as arbiters to ensure that all concurrent transactions
+ * use the same set of arbiters.
+ */
+ if (nonAncestorIdxs)
+ {
+ for (i = 0; i < leaf_part_rri->ri_NumIndices; i++)
+ {
+ if (list_member_oid(nonAncestorIdxs, leaf_part_rri->ri_IndexRelationDescs[i]->rd_index->indexrelid))
+ {
+ Relation nonAncestorIndexRelation = leaf_part_rri->ri_IndexRelationDescs[i];
+ IndexInfo *nonAncestorIndexInfo = leaf_part_rri->ri_IndexRelationInfo[i];
+ Assert(!list_member_oid(arbiterIndexes, nonAncestorIndexRelation->rd_index->indexrelid));
+
+ /* It is too early to us non-ready indexes as arbiters */
+ if (!nonAncestorIndexInfo->ii_ReadyForInserts)
+ continue;
+
+ for (j = 0; j < leaf_part_rri->ri_NumIndices; j++)
+ {
+ if (list_member_oid(arbiterIndexes,
+ leaf_part_rri->ri_IndexRelationDescs[j]->rd_index->indexrelid))
+ {
+ Relation arbiterIndexRelation = leaf_part_rri->ri_IndexRelationDescs[j];
+ IndexInfo *arbiterIndexInfo = leaf_part_rri->ri_IndexRelationInfo[j];
+
+ /* If non-ancestor index are compatible to arbiter - use it as arbiter too. */
+ if (IsIndexCompatibleAsArbiter(arbiterIndexRelation, arbiterIndexInfo,
+ nonAncestorIndexRelation, nonAncestorIndexInfo))
+ {
+ arbiterIndexes = lappend_oid(arbiterIndexes,
+ nonAncestorIndexRelation->rd_index->indexrelid);
+ additional_arbiters++;
+ }
+ }
+ }
+ }
+ }
+ }
+ list_free(nonAncestorIdxs);
+
+ /*
+ * If the resulting lists are of inequal length, something is wrong.
+ * (This shouldn't happen, since arbiter index selection should not
+ * pick up a non-ready index.)
+ *
+ * But we need to consider an additional arbiter indexes also.
+ */
+ if (list_length(rootResultRelInfo->ri_onConflictArbiterIndexes) !=
+ list_length(arbiterIndexes) - additional_arbiters)
+ elog(ERROR, "invalid arbiter index list");
}
-
- /*
- * If the resulting lists are of inequal length, something is wrong.
- * (This shouldn't happen, since arbiter index selection should not
- * pick up an invalid index.)
- */
- if (list_length(rootResultRelInfo->ri_onConflictArbiterIndexes) !=
- list_length(arbiterIndexes))
- elog(ERROR, "invalid arbiter index list");
leaf_part_rri->ri_onConflictArbiterIndexes = arbiterIndexes;
/*
--
2.34.1
stress_test.__patch__
Description: Binary data
