0009-Extend-the-concurrent-abort-handling-for-in-progress.patch

application/octet-stream

Filename: 0009-Extend-the-concurrent-abort-handling-for-in-progress.patch
Type: application/octet-stream
Part: 9
Message: Re: PATCH: logical_work_mem and logical streaming of large in-progress transactions

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: format-patch
Series: patch 0009
Subject: Extend the concurrent abort handling for in-progress transaction
File+
src/backend/replication/logical/reorderbuffer.c 19 1
From 2a0712fb830a6be20296d18fb821862e302aeb8b Mon Sep 17 00:00:00 2001
From: Dilip Kumar <dilip.kumar@enterprisedb.com>
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