mirrors/postgres
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Rearrange logrep worker's snapshot handling some more.

Browse Source 
It turns out that worker.c's code path for TRUNCATE was also
careless about establishing a snapshot while executing user-defined
code, allowing the checks added by commit 84f5c2908 to fail when
a trigger is fired in that context.

We could just wrap Push/PopActiveSnapshot around the truncate call,
but it seems better to establish a policy of holding a snapshot
throughout execution of a replication step.  To help with that and
possible future requirements, replace the previous ensure_transaction
calls with pairs of begin/end_replication_step calls.

Per report from Mark Dilger.  Back-patch to v11, like the previous
changes.

Discussion: https://postgr.es/m/B4A3AF82-79ED-4F4C-A4E5-CD2622098972@enterprisedb.com
This commit is contained in:
Tom Lane 2021-06-10 12:27:27 -04:00
parent 2208d71a00
commit 26383da7d4
1 changed files with 38 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
src/backend/replication/logical/worker.c
View File

@ -138,30 +138,41 @@ should_apply_changes_for_rel(LogicalRepRelMapEntry *rel)
}
/*
* Make sure that we started local transaction.
* Begin one step (one INSERT, UPDATE, etc) of a replication transaction.
*
* Also switches to ApplyMessageContext as necessary.
* Start a transaction, if this is the first step (else we keep using the
* existing transaction).
* Also provide a global snapshot and ensure we run in ApplyMessageContext.
*/
static bool
ensure_transaction(void)
{
if (IsTransactionState())
static void
begin_replication_step(void)
{
SetCurrentStatementStartTimestamp();
if (CurrentMemoryContext != ApplyMessageContext)
MemoryContextSwitchTo(ApplyMessageContext);
return false;
if (!IsTransactionState())
{
StartTransactionCommand();
maybe_reread_subscription();
}
SetCurrentStatementStartTimestamp();
StartTransactionCommand();
maybe_reread_subscription();
PushActiveSnapshot(GetTransactionSnapshot());
MemoryContextSwitchTo(ApplyMessageContext);
return true;
}
/*
* Finish up one step of a replication transaction.
* Callers of begin_replication_step() must also call this.
*
* We don't close out the transaction here, but we should increment
* the command counter to make the effects of this step visible.
*/
static void
end_replication_step(void)
{
PopActiveSnapshot();
CommandCounterIncrement();
}
@ -178,13 +189,6 @@ create_estate_for_relation(LogicalRepRelMapEntry *rel)
ResultRelInfo *resultRelInfo;
RangeTblEntry *rte;
/*
* Input functions may need an active snapshot, as may AFTER triggers
* invoked during finish_estate. For safety, ensure an active snapshot
* exists throughout all our usage of the executor.
*/
PushActiveSnapshot(GetTransactionSnapshot());
estate = CreateExecutorState();
rte = makeNode(RangeTblEntry);
@ -222,7 +226,6 @@ finish_estate(EState *estate)
/* Cleanup. */
ExecResetTupleTable(estate->es_tupleTable, false);
FreeExecutorState(estate);
PopActiveSnapshot();
}
/*
@ -612,7 +615,7 @@ apply_handle_insert(StringInfo s)
TupleTableSlot *remoteslot;
MemoryContext oldctx;
ensure_transaction();
begin_replication_step();
relid = logicalrep_read_insert(s, &newtup);
rel = logicalrep_rel_open(relid, RowExclusiveLock);
@ -623,6 +626,7 @@ apply_handle_insert(StringInfo s)
* transaction so it's safe to unlock it.
*/
logicalrep_rel_close(rel, RowExclusiveLock);
end_replication_step();
return;
}
@ -650,7 +654,7 @@ apply_handle_insert(StringInfo s)
logicalrep_rel_close(rel, NoLock);
CommandCounterIncrement();
end_replication_step();
}
/*
@ -708,7 +712,7 @@ apply_handle_update(StringInfo s)
bool found;
MemoryContext oldctx;
ensure_transaction();
begin_replication_step();
relid = logicalrep_read_update(s, &has_oldtup, &oldtup,
&newtup);
@ -720,6 +724,7 @@ apply_handle_update(StringInfo s)
* transaction so it's safe to unlock it.
*/
logicalrep_rel_close(rel, RowExclusiveLock);
end_replication_step();
return;
}
@ -825,7 +830,7 @@ apply_handle_update(StringInfo s)
logicalrep_rel_close(rel, NoLock);
CommandCounterIncrement();
end_replication_step();
}
/*
@ -847,7 +852,7 @@ apply_handle_delete(StringInfo s)
bool found;
MemoryContext oldctx;
ensure_transaction();
begin_replication_step();
relid = logicalrep_read_delete(s, &oldtup);
rel = logicalrep_rel_open(relid, RowExclusiveLock);
@ -858,6 +863,7 @@ apply_handle_delete(StringInfo s)
* transaction so it's safe to unlock it.
*/
logicalrep_rel_close(rel, RowExclusiveLock);
end_replication_step();
return;
}
@ -920,7 +926,7 @@ apply_handle_delete(StringInfo s)
logicalrep_rel_close(rel, NoLock);
CommandCounterIncrement();
end_replication_step();
}
/*
@ -941,7 +947,7 @@ apply_handle_truncate(StringInfo s)
ListCell *lc;
LOCKMODE lockmode = AccessExclusiveLock;
ensure_transaction();
begin_replication_step();
remote_relids = logicalrep_read_truncate(s, &cascade, &restart_seqs);
@ -982,7 +988,7 @@ apply_handle_truncate(StringInfo s)
logicalrep_rel_close(rel, NoLock);
}
CommandCounterIncrement();
end_replication_step();
}