From 2a0712fb830a6be20296d18fb821862e302aeb8b Mon Sep 17 00:00:00 2001 From: Dilip Kumar Date: Thu, 3 Oct 2019 09:04:09 +0530 Subject: [PATCH 09/13] Extend the concurrent abort handling for in-progress transaction --- src/backend/replication/logical/reorderbuffer.c | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index c2d4677..963cbf3 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -3280,6 +3280,7 @@ ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn) volatile CommandId command_id; bool using_subtxn; Size streamed = 0; + MemoryContext ccxt = CurrentMemoryContext; ReorderBufferStreamIterTXNState *volatile iterstate = NULL; /* @@ -3410,6 +3411,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: @@ -3722,6 +3730,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); @@ -3740,7 +3751,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(); -- 1.8.3.1