diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index fd9efd94737..dd42acc13e2 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -177,9 +177,14 @@ 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, + bool skipduplicates, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot) @@ -192,13 +197,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. */ @@ -220,7 +222,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_array(TypeCacheEntry *, outslot->tts_tupleDescriptor->natts); @@ -629,9 +631,12 @@ RelationFindDeletedTupleInfoSeq(Relation rel, TupleTableSlot *searchslot, /* * Similar to RelationFindDeletedTupleInfoSeq() but using index scan to locate * the deleted tuple. + * + * 'skipduplicates' works as in RelationFindReplTupleByIndex(). */ bool RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid, + bool skipduplicates, TupleTableSlot *searchslot, TransactionId oldestxmin, TransactionId *delete_xid, @@ -644,7 +649,6 @@ RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid, IndexScanDesc scan; TupleTableSlot *scanslot; TypeCacheEntry **eq = NULL; - bool isIdxSafeToSkipDuplicates; TupleDesc desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(rel); Assert(equalTupleDescs(desc, searchslot->tts_tupleDescriptor)); @@ -654,8 +658,6 @@ RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid, *delete_time = 0; *delete_origin = InvalidReplOriginId; - isIdxSafeToSkipDuplicates = (GetRelationIdentityOrPK(rel) == idxoid); - scanslot = table_slot_create(rel, NULL); idxrel = index_open(idxoid, RowExclusiveLock); @@ -681,7 +683,7 @@ RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid, * Avoid expensive equality check if the index is primary key or * replica identity index. */ - if (!isIdxSafeToSkipDuplicates) + if (!skipduplicates) { if (eq == NULL) eq = palloc0_array(TypeCacheEntry *, scanslot->tts_tupleDescriptor->natts); diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c index b71a0c9e206..90431ffce98 100644 --- a/src/backend/replication/logical/conflict.c +++ b/src/backend/replication/logical/conflict.c @@ -743,6 +743,10 @@ get_tuple_desc(EState *estate, ResultRelInfo *relinfo, ConflictType type, * 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 8ffd2583afb..6242ce70ad2 100644 --- a/src/backend/replication/logical/relation.c +++ b/src/backend/replication/logical/relation.c @@ -56,7 +56,7 @@ typedef struct LogicalRepPartMapEntry } LogicalRepPartMapEntry; static Oid FindLogicalRepLocalIndex(Relation localrel, LogicalRepRelation *remoterel, - AttrMap *attrMap); + AttrMap *attrMap, bool *idxisreplident); /* * Relcache invalidation callback for our relation map cache. @@ -497,7 +497,8 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode) * on the relation). */ entry->localindexoid = FindLogicalRepLocalIndex(entry->localrel, remoterel, - entry->attrmap); + entry->attrmap, + &entry->idxisreplident); entry->localrelvalid = true; } @@ -764,7 +765,8 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root, * anything in the LogicalRepPartMapContext (hence CacheMemoryContext). */ entry->localindexoid = FindLogicalRepLocalIndex(partrel, remoterel, - entry->attrmap); + entry->attrmap, + &entry->idxisreplident); entry->localrelvalid = true; @@ -925,13 +927,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. @@ -944,7 +951,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 7781bb1c168..ab7c4ced66d 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -290,6 +290,7 @@ #include "tcop/tcopprot.h" #include "utils/acl.h" #include "utils/guc.h" +#include "utils/injection_point.h" #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -609,19 +610,17 @@ 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 bool FindDeletedTupleInLocalRel(Relation localrel, - Oid localidxoid, + LogicalRepRelMapEntry *relmapentry, TupleTableSlot *remoteslot, TransactionId *delete_xid, ReplOriginId *delete_origin, @@ -2800,11 +2799,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), @@ -2931,7 +2927,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); @@ -2955,8 +2951,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; @@ -2968,11 +2963,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); /* @@ -3026,7 +3022,7 @@ apply_handle_update_internal(ApplyExecutionData *edata, * Detecting whether the tuple was recently deleted or never existed * is crucial to avoid misleading the user during conflict handling. */ - if (FindDeletedTupleInLocalRel(localrel, localindexoid, remoteslot, + if (FindDeletedTupleInLocalRel(localrel, relmapentry, remoteslot, &conflicttuple.xmin, &conflicttuple.origin, &conflicttuple.ts) && @@ -3128,7 +3124,7 @@ apply_handle_delete(StringInfo s) ExecOpenIndices(relinfo, false); apply_handle_delete_internal(edata, relinfo, - remoteslot, rel->localindexoid); + remoteslot, rel); ExecCloseIndices(relinfo); } @@ -3154,11 +3150,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}; @@ -3171,7 +3166,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. */ @@ -3216,16 +3211,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; /* @@ -3237,22 +3236,39 @@ 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, + relmapentry->idxisreplident, LockTupleExclusive, remoteslot, *localslot); } @@ -3309,16 +3325,21 @@ IsIndexUsableForFindingDeletedTuple(Oid localindexoid, * The search is performed using either the replica identity index, primary * key, other available index, or a sequential scan if necessary. * + * 'relmapentry' is the relation map entry for 'localrel', as in + * FindReplTupleInLocalRel(). + * * Returns true if the deleted tuple is found. If found, the transaction ID, * origin, and commit timestamp of the deletion are stored in '*delete_xid', * '*delete_origin', and '*delete_time' respectively. */ static bool -FindDeletedTupleInLocalRel(Relation localrel, Oid localidxoid, +FindDeletedTupleInLocalRel(Relation localrel, + LogicalRepRelMapEntry *relmapentry, TupleTableSlot *remoteslot, TransactionId *delete_xid, ReplOriginId *delete_origin, TimestampTz *delete_time) { + Oid localidxoid = relmapentry->localindexoid; TransactionId oldestxmin; /* @@ -3383,6 +3404,7 @@ FindDeletedTupleInLocalRel(Relation localrel, Oid localidxoid, if (OidIsValid(localidxoid) && IsIndexUsableForFindingDeletedTuple(localidxoid, oldestxmin)) return RelationFindDeletedTupleInfoByIndex(localrel, localidxoid, + relmapentry->idxisreplident, remoteslot, oldestxmin, delete_xid, delete_origin, delete_time); @@ -3484,8 +3506,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: @@ -3505,9 +3526,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) { @@ -3519,8 +3538,7 @@ apply_handle_tuple_routing(ApplyExecutionData *edata, * never existed is crucial to avoid misleading the user * during conflict handling. */ - if (FindDeletedTupleInLocalRel(partrel, - part_entry->localindexoid, + if (FindDeletedTupleInLocalRel(partrel, part_entry, remoteslot_part, &conflicttuple.xmin, &conflicttuple.origin, diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index 8bb6c7bda2f..23a09a70aa2 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -775,6 +775,7 @@ extern void check_exclusion_constraint(Relation heap, Relation index, * prototypes from functions in execReplication.c */ extern bool RelationFindReplTupleByIndex(Relation rel, Oid idxoid, + bool skipduplicates, LockTupleMode lockmode, TupleTableSlot *searchslot, TupleTableSlot *outslot); @@ -787,6 +788,7 @@ extern bool RelationFindDeletedTupleInfoSeq(Relation rel, ReplOriginId *delete_origin, TimestampTz *delete_time); extern bool RelationFindDeletedTupleInfoByIndex(Relation rel, Oid idxoid, + bool skipduplicates, TupleTableSlot *searchslot, TransactionId oldestxmin, TransactionId *delete_xid, diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h index efe0f9d6031..4063118d2b9 100644 --- a/src/include/replication/logicalrelation.h +++ b/src/include/replication/logicalrelation.h @@ -33,6 +33,10 @@ typedef struct LogicalRepRelMapEntry AttrMap *attrmap; /* map of local attributes to remote ones */ bool updatable; /* Can apply updates/deletes? */ Oid localindexoid; /* which index to use, or InvalidOid if none */ + bool idxisreplident; /* is it the relation's replica identity or + * primary key, rather than an index usable + * for a REPLICA IDENTITY FULL remote + * relation? */ /* Sync state. */ char state; diff --git a/src/test/subscription/t/032_subscribe_use_index.pl b/src/test/subscription/t/032_subscribe_use_index.pl index 1ccd36ac227..bd80465f727 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');