diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index bb78d89265d..c21c8e720ad 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -174,12 +174,17 @@ should_refetch_tuple(TM_Result res, TM_FailureData *tmfd) * * If a matching tuple is found, lock it with lockmode, fill the slot with its * contents, and return true. Return false otherwise. + * + * 'skipduplicates' specifies whether the first matching tuple can be used + * without comparing it against 'searchslot'. If false, all matching tuples are + * compared against 'searchslot', which must contain a complete row. */ bool -RelationFindReplTupleByIndex(Relation rel, Oid idxoid, - LockTupleMode lockmode, - TupleTableSlot *searchslot, - TupleTableSlot *outslot) +RelationFindReplTupleByIndexExt(Relation rel, Oid idxoid, + bool skipduplicates, + LockTupleMode lockmode, + TupleTableSlot *searchslot, + TupleTableSlot *outslot) { ScanKeyData skey[INDEX_MAX_KEYS]; int skey_attoff; @@ -189,13 +194,10 @@ RelationFindReplTupleByIndex(Relation rel, Oid idxoid, Relation idxrel; bool found; TypeCacheEntry **eq = NULL; - bool isIdxSafeToSkipDuplicates; /* Open the index. */ idxrel = index_open(idxoid, RowExclusiveLock); - isIdxSafeToSkipDuplicates = (GetRelationIdentityOrPK(rel) == idxoid); - InitDirtySnapshot(snap); /* Build scan key. */ @@ -216,7 +218,7 @@ retry: * Avoid expensive equality check if the index is primary key or * replica identity index. */ - if (!isIdxSafeToSkipDuplicates) + if (!skipduplicates) { if (eq == NULL) eq = palloc0(sizeof(*eq) * outslot->tts_tupleDescriptor->natts); @@ -275,6 +277,25 @@ retry: return found; } +/* + * ABI-compatible wrapper to emulate old version of the above function. + * Do not call this version in new code. + */ +bool +RelationFindReplTupleByIndex(Relation rel, Oid idxoid, + LockTupleMode lockmode, + TupleTableSlot *searchslot, + TupleTableSlot *outslot) +{ + bool skipduplicates; + + skipduplicates = (GetRelationIdentityOrPK(rel) == idxoid); + + return RelationFindReplTupleByIndexExt(rel, idxoid, + skipduplicates, + lockmode, searchslot, outslot); +} + /* * Compare the tuples in the slots by checking if they have equal values. */ diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index 6f760f9a135..4ffc83f5927 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -410,6 +410,10 @@ build_tuple_value_details(EState *estate, ResultRelInfo *relinfo, * when applying update or delete, such an index scan may not result * in a unique tuple and we still compare the complete tuple in such * cases, thus such indexes are not used here. + * + * XXX This can disagree with the index the apply worker searched by, + * see FindReplTupleInLocalRel(). It may not even be one that + * ExecOpenIndices() locked. */ Oid replica_index = GetRelationIdentityOrPK(localrel); diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c index 0915b4d8b15..b605e97c301 100644 --- a/src/backend/replication/logical/relation.c +++ b/src/backend/replication/logical/relation.c @@ -55,7 +55,7 @@ typedef struct LogicalRepPartMapEntry } LogicalRepPartMapEntry; static Oid FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel, - AttrMap *attrMap); + AttrMap *attrMap, bool *idxisreplident); /* * Relcache invalidation callback for our relation map cache. @@ -484,7 +484,8 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode) * on the relation). */ entry->localindexoid = FindLogicalRepLocalIndex(entry->localrel, remoterel, - entry->attrmap); + entry->attrmap, + &entry->idxisreplident); entry->localrelvalid = true; } @@ -756,7 +757,8 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root, * anything in the LogicalRepPartMapContext (hence CacheMemoryContext). */ entry->localindexoid = FindLogicalRepLocalIndex(partrel, remoterel, - entry->attrmap); + entry->attrmap, + &entry->idxisreplident); entry->localrelvalid = true; @@ -917,13 +919,18 @@ GetRelationIdentityOrPK(Relation rel) /* * Returns the index oid if we can use an index for subscriber. Otherwise, * returns InvalidOid. + * + * '*idxisreplident' is true if the returned index is the relation's replica + * identity or primary key, and false otherwise. */ static Oid FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel, - AttrMap *attrMap) + AttrMap *attrMap, bool *idxisreplident) { Oid idxoid; + *idxisreplident = false; + /* * We never need index oid for partitioned tables, always rely on leaf * partition's index. @@ -936,7 +943,10 @@ FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel, */ idxoid = GetRelationIdentityOrPK(localrel); if (OidIsValid(idxoid)) + { + *idxisreplident = true; return idxoid; + } if (remoterel->replident == REPLICA_IDENTITY_FULL) { diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 5911c4743ab..bc4049f4d9c 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -176,6 +176,7 @@ #include "utils/acl.h" #include "utils/dynahash.h" #include "utils/guc.h" +#include "utils/injection_point.h" #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -379,15 +380,13 @@ static void apply_handle_insert_internal(ApplyExecutionData *edata, static void apply_handle_update_internal(ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, - LogicalRepTupleData *newtup, - Oid localindexoid); + LogicalRepTupleData *newtup); static void apply_handle_delete_internal(ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, - Oid localindexoid); + LogicalRepRelMapEntry *relmapentry); static bool FindReplTupleInLocalRel(ApplyExecutionData *edata, Relation localrel, - LogicalRepRelation *remoterel, - Oid localidxoid, + LogicalRepRelMapEntry *relmapentry, TupleTableSlot *remoteslot, TupleTableSlot **localslot); static void apply_handle_tuple_routing(ApplyExecutionData *edata, @@ -2532,11 +2531,8 @@ check_relation_updatable(LogicalRepRelMapEntry *rel) if (rel->updatable) return; - /* - * We are in error mode so it's fine this is somewhat slow. It's better to - * give user correct error. - */ - if (OidIsValid(GetRelationIdentityOrPK(rel->localrel))) + /* Use the entry, so this matches what updatable was decided from. */ + if (rel->idxisreplident) { ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), @@ -2663,7 +2659,7 @@ apply_handle_update(StringInfo s) remoteslot, &newtup, CMD_UPDATE); else apply_handle_update_internal(edata, edata->targetRelInfo, - remoteslot, &newtup, rel->localindexoid); + remoteslot, &newtup); finish_edata(edata); @@ -2687,8 +2683,7 @@ static void apply_handle_update_internal(ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, - LogicalRepTupleData *newtup, - Oid localindexoid) + LogicalRepTupleData *newtup) { EState *estate = edata->estate; LogicalRepRelMapEntry *relmapentry = edata->targetRel; @@ -2700,11 +2695,12 @@ apply_handle_update_internal(ApplyExecutionData *edata, MemoryContext oldctx; EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1, NIL); + + INJECTION_POINT("apply-update-before-open-indices", NULL); + ExecOpenIndices(relinfo, false); - found = FindReplTupleInLocalRel(edata, localrel, - &relmapentry->remoterel, - localindexoid, + found = FindReplTupleInLocalRel(edata, localrel, relmapentry, remoteslot, &localslot); /* @@ -2846,7 +2842,7 @@ apply_handle_delete(StringInfo s) ExecOpenIndices(relinfo, false); apply_handle_delete_internal(edata, relinfo, - remoteslot, rel->localindexoid); + remoteslot, rel); ExecCloseIndices(relinfo); } @@ -2872,11 +2868,10 @@ static void apply_handle_delete_internal(ApplyExecutionData *edata, ResultRelInfo *relinfo, TupleTableSlot *remoteslot, - Oid localindexoid) + LogicalRepRelMapEntry *relmapentry) { EState *estate = edata->estate; Relation localrel = relinfo->ri_RelationDesc; - LogicalRepRelation *remoterel = &edata->targetRel->remoterel; EPQState epqstate; TupleTableSlot *localslot; ConflictTupleInfo conflicttuple = {0}; @@ -2889,7 +2884,7 @@ apply_handle_delete_internal(ApplyExecutionData *edata, !localrel->rd_rel->relhasindex || RelationGetIndexList(localrel) == NIL); - found = FindReplTupleInLocalRel(edata, localrel, remoterel, localindexoid, + found = FindReplTupleInLocalRel(edata, localrel, relmapentry, remoteslot, &localslot); /* If found delete it. */ @@ -2934,16 +2929,20 @@ apply_handle_delete_internal(ApplyExecutionData *edata, * the corresponding local relation using either replica identity index, * primary key, index or if needed, sequential scan. * + * 'relmapentry' is the relation map entry for 'localrel'. It tells which + * index to use, if any, and whether that index is the relation's replica + * identity or primary key. + * * Local tuple, if found, is returned in '*localslot'. */ static bool FindReplTupleInLocalRel(ApplyExecutionData *edata, Relation localrel, - LogicalRepRelation *remoterel, - Oid localidxoid, + LogicalRepRelMapEntry *relmapentry, TupleTableSlot *remoteslot, TupleTableSlot **localslot) { EState *estate = edata->estate; + Oid localidxoid = relmapentry->localindexoid; bool found; /* @@ -2955,24 +2954,41 @@ FindReplTupleInLocalRel(ApplyExecutionData *edata, Relation localrel, *localslot = table_slot_create(localrel, &estate->es_tupleTable); Assert(OidIsValid(localidxoid) || - (remoterel->replident == REPLICA_IDENTITY_FULL)); + (relmapentry->remoterel.replident == REPLICA_IDENTITY_FULL)); if (OidIsValid(localidxoid)) { #ifdef USE_ASSERT_CHECKING Relation idxrel = index_open(localidxoid, AccessShareLock); - /* Index must be PK, RI, or usable for REPLICA IDENTITY FULL tables */ - Assert(GetRelationIdentityOrPK(localrel) == localidxoid || - (remoterel->replident == REPLICA_IDENTITY_FULL && - IsIndexUsableForReplicaIdentityFull(idxrel, - edata->targetRel->attrmap))); + if (relmapentry->idxisreplident) + { + /* + * We cannot assert this is still the replica identity or primary + * key. DROP INDEX CONCURRENTLY and REINDEX CONCURRENTLY clear + * indisvalid and indisreplident without conflicting with our + * RowExclusiveLock, so GetRelationIdentityOrPK() may no longer + * return it. Unique and non-partial is what the scan actually + * relies on, and no DDL can take those away. + */ + Assert(idxrel->rd_index->indisunique); + Assert(heap_attisnull(idxrel->rd_indextuple, + Anum_pg_index_indpred, NULL)); + } + else + { + /* Otherwise every match is compared, so we need a whole row. */ + Assert(relmapentry->remoterel.replident == REPLICA_IDENTITY_FULL); + Assert(IsIndexUsableForReplicaIdentityFull(idxrel, + relmapentry->attrmap)); + } index_close(idxrel, AccessShareLock); #endif - found = RelationFindReplTupleByIndex(localrel, localidxoid, - LockTupleExclusive, - remoteslot, *localslot); + found = RelationFindReplTupleByIndexExt(localrel, localidxoid, + relmapentry->idxisreplident, + LockTupleExclusive, + remoteslot, *localslot); } else found = RelationFindReplTupleSeq(localrel, LockTupleExclusive, @@ -3072,8 +3088,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, case CMD_DELETE: apply_handle_delete_internal(edata, partrelinfo, - remoteslot_part, - part_entry->localindexoid); + remoteslot_part, part_entry); break; case CMD_UPDATE: @@ -3093,9 +3108,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, ConflictTupleInfo conflicttuple = {0}; /* Get the matching local tuple from the partition. */ - found = FindReplTupleInLocalRel(edata, partrel, - &part_entry->remoterel, - part_entry->localindexoid, + found = FindReplTupleInLocalRel(edata, partrel, part_entry, remoteslot_part, &localslot); if (!found) { diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index b6697c0fc75..ae4181e20eb 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -760,6 +760,11 @@ extern bool RelationFindReplTupleByIndex(Relation rel, Oid idxoid, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot); +extern bool RelationFindReplTupleByIndexExt(Relation rel, Oid idxoid, + bool skipduplicates, + LockTupleMode lockmode, + TupleTableSlot *searchslot, + TupleTableSlot *outslot); extern bool RelationFindReplTupleSeq(Relation rel, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot); diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h index 7a561a8e8d8..8362f8f064e 100644 --- a/src/include/replication/logicalrelation.h +++ b/src/include/replication/logicalrelation.h @@ -32,6 +32,10 @@ typedef struct LogicalRepRelMapEntry Relation localrel; /* relcache entry (NULL when closed) */ AttrMap *attrmap; /* map of local attributes to remote ones */ bool updatable; /* Can apply updates/deletes? */ + bool idxisreplident; /* is localindexoid the relation's replica + * identity or primary key, rather than an + * index usable for a REPLICA IDENTITY FULL + * remote relation? */ Oid localindexoid; /* which index to use, or InvalidOid if none */ /* Sync state. */ diff --git a/src/test/subscription/Makefile b/src/test/subscription/Makefile index 50b65d8f6ea..9d97e7d5c0d 100644 --- a/src/test/subscription/Makefile +++ b/src/test/subscription/Makefile @@ -13,9 +13,11 @@ subdir = src/test/subscription top_builddir = ../../.. include $(top_builddir)/src/Makefile.global -EXTRA_INSTALL = contrib/hstore +EXTRA_INSTALL = contrib/hstore \ + src/test/modules/injection_points export with_icu +export enable_injection_points check: $(prove_check) diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build index 586ffba434e..20b4e523d93 100644 --- a/src/test/subscription/meson.build +++ b/src/test/subscription/meson.build @@ -5,7 +5,10 @@ tests += { 'sd': meson.current_source_dir(), 'bd': meson.current_build_dir(), 'tap': { - 'env': {'with_icu': icu.found() ? 'yes' : 'no'}, + 'env': { + 'with_icu': icu.found() ? 'yes' : 'no', + 'enable_injection_points': get_option('injection_points') ? 'yes' : 'no', + }, 'tests': [ 't/001_rep_changes.pl', 't/002_types.pl', diff --git a/src/test/subscription/t/032_subscribe_use_index.pl b/src/test/subscription/t/032_subscribe_use_index.pl index 6c6cfdae38a..28479f3a039 100644 --- a/src/test/subscription/t/032_subscribe_use_index.pl +++ b/src/test/subscription/t/032_subscribe_use_index.pl @@ -606,6 +606,133 @@ $node_subscriber->safe_psql('postgres', "DROP TABLE test_replica_id_full"); # Testcase end: Subscription can use hash index # ============================================================================= +# ============================================================================= +# Testcase start: Subscription keeps using an index that concurrent DDL has +# demoted from replica identity +# +# DROP INDEX CONCURRENTLY clears indisvalid and indisreplident and commits that +# before waiting for the lock the apply worker holds, so the worker can be left +# holding an index the catalogs no longer call the replica identity. The drop +# gets no further while apply holds the table, so the index is still complete +# and still maintained, and the change must be applied through it rather than +# dropped as a missing-tuple conflict. +# +# REINDEX CONCURRENTLY reaches the same state by swapping a new index in, but +# the apply path is the same one, so it is not tested separately. + +SKIP: +{ + skip 'Injection points not supported by this build', 5 + unless $ENV{enable_injection_points} eq 'yes'; + skip 'Extension injection_points not installed', 5 + unless $node_subscriber->check_extension('injection_points'); + + $node_subscriber->safe_psql('postgres', + 'CREATE EXTENSION injection_points'); + + # create tables pub and sub, using a unique index as replica identity + $node_publisher->safe_psql( + 'postgres', q[ + CREATE TABLE test_dropri (x int NOT NULL, y int); + CREATE UNIQUE INDEX test_dropri_ri ON test_dropri (x); + ALTER TABLE test_dropri REPLICA IDENTITY USING INDEX test_dropri_ri; + INSERT INTO test_dropri SELECT i, i FROM generate_series(1,20) i; + CREATE PUBLICATION tap_pub_dropri FOR TABLE test_dropri; + ]); + $node_subscriber->safe_psql( + 'postgres', q[ + CREATE TABLE test_dropri (x int NOT NULL, y int); + CREATE UNIQUE INDEX test_dropri_ri ON test_dropri (x); + ALTER TABLE test_dropri REPLICA IDENTITY USING INDEX test_dropri_ri; + ]); + $node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub_dropri CONNECTION '$publisher_connstr application_name=dropri' PUBLICATION tap_pub_dropri" + ); + + # wait for initial table synchronization to finish + $node_subscriber->wait_for_subscription_sync($node_publisher, 'dropri'); + + # Let the worker take the index and stop before opening the relation's + # indexes. The point is attached server-wide: the apply worker is not a + # session this test can attach anything in. + $node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('apply-update-before-open-indices', 'wait')" + ); + $node_publisher->safe_psql('postgres', + "UPDATE test_dropri SET y = 99 WHERE x = 7"); + $node_subscriber->wait_for_event( + 'logical replication apply worker', + 'apply-update-before-open-indices'); + + # This commits the loss of the replica identity, leaving relreplident set + # to 'i' with no index claiming to be that identity, then parks waiting + # for the apply worker's lock on the table. + my $log_offset = -s $node_subscriber->logfile; + my $drop = $node_subscriber->background_psql('postgres'); + $drop->query_until( + qr/starting_drop/, q[ + \echo starting_drop + DROP INDEX CONCURRENTLY test_dropri_ri; + ]); + $node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_index" + . " WHERE indrelid = 'test_dropri'::regclass AND indisreplident") + or die "timed out waiting for the identity index to be invalidated"; + + # Detach before waking, so the worker cannot park on the point again. + $node_subscriber->safe_psql( + 'postgres', + "SELECT injection_points_detach('apply-update-before-open-indices'); + SELECT injection_points_wakeup('apply-update-before-open-indices');" + ); + + # The straddling change goes through, found by the demoted index. + $node_publisher->wait_for_catchup('dropri'); + $result = $node_subscriber->safe_psql('postgres', + "SELECT y FROM test_dropri WHERE x = 7"); + is($result, qq(99), 'change straddling the drop is applied'); + ok($drop->quit, 'DROP INDEX CONCURRENTLY completes'); + + # From the next change on, the relation map entry is rebuilt, finds no + # replica identity, and apply stops with the usual error. + $node_publisher->safe_psql('postgres', + "UPDATE test_dropri SET y = 123 WHERE x = 8"); + ok( $node_subscriber->poll_query_until( + 'postgres', q[ + SELECT apply_error_count > 0 FROM pg_stat_subscription_stats + WHERE subname = 'tap_sub_dropri']), + 'later changes wait for a replica identity'); + like( + slurp_file($node_subscriber->logfile, $log_offset), + qr/logical replication target relation "public\.test_dropri" has neither REPLICA IDENTITY index nor PRIMARY KEY/, + 'and say why'); + + # Give the relation a replica identity again and they resume. + $node_subscriber->safe_psql( + 'postgres', q[ + CREATE UNIQUE INDEX test_dropri_ri2 ON test_dropri (x); + ALTER TABLE test_dropri REPLICA IDENTITY USING INDEX test_dropri_ri2; + ]); + $node_publisher->wait_for_catchup('dropri'); + $result = $node_subscriber->safe_psql('postgres', + "SELECT y FROM test_dropri WHERE x = 8"); + is($result, qq(123), 'replication resumes'); + + # cleanup pub + $node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_dropri"); + $node_publisher->safe_psql('postgres', "DROP TABLE test_dropri"); + # cleanup sub + $node_subscriber->safe_psql('postgres', + "DROP SUBSCRIPTION tap_sub_dropri"); + $node_subscriber->safe_psql('postgres', "DROP TABLE test_dropri"); + $node_subscriber->safe_psql('postgres', + "DROP EXTENSION injection_points"); +} + +# Testcase end: Subscription keeps using an index that concurrent DDL has +# demoted from replica identity +# ============================================================================= + $node_subscriber->stop('fast'); $node_publisher->stop('fast');