handle_concurrent_abort_for_in_progress_transaction.patch
application/octet-stream
Filename: handle_concurrent_abort_for_in_progress_transaction.patch
Type: application/octet-stream
Part: 0
Patch
Same data as JSON:
GET /api/v1/attachments/:id/patch
the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes.
API reference →
Format: unified
| File | + | − |
|---|---|---|
| src/backend/replication/logical/reorderbuffer.c | 19 | 1 |
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 163395f..ca7e2d7 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -3449,6 +3449,7 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
volatile CommandId command_id;
bool using_subtxn;
Size streamed = 0;
+ MemoryContext ccxt = CurrentMemoryContext;
ReorderBufferStreamIterTXNState *volatile iterstate = NULL;
/*
@@ -3579,6 +3580,13 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
/* we're going to stream this change */
streamed++;
+ /*
+ * Set the CheckXidAlive to the current (sub)xid for which this
+ * change belongs to so that we can detect the abort while we are
+ * decoding.
+ */
+ CheckXidAlive = change->txn->xid;
+
switch (change->action)
{
case REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM:
@@ -3891,6 +3899,9 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
}
PG_CATCH();
{
+ MemoryContext ecxt = MemoryContextSwitchTo(ccxt);
+ ErrorData *errdata = CopyErrorData();
+
/* TODO: Encapsulate cleanup from the PG_TRY and PG_CATCH blocks */
if (iterstate)
ReorderBufferStreamIterTXNFinish(rb, iterstate);
@@ -3909,7 +3920,14 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
if (using_subtxn)
RollbackAndReleaseCurrentSubTransaction();
- PG_RE_THROW();
+ /* re-throw only if it's not an abort */
+ if (errdata->sqlerrcode != ERRCODE_TRANSACTION_ROLLBACK)
+ {
+ MemoryContextSwitchTo(ecxt);
+ PG_RE_THROW();
+ }
+ else
+ FlushErrorState();
}
PG_END_TRY();