postpone ReorderBuffer cleanup

kelvich · kelvich · commit 942d9165d268 · 2017-03-28T03:38:31.000+03:00
diff --git a/contrib/test_decoding/expected/prepared.out b/contrib/test_decoding/expected/prepared.out
@@ -98,7 +98,7 @@ SELECT data FROM pg_logical_slot_peek_changes('regression_slot', NULL, NULL, 'in
  BEGIN
  table public.test_prepared2: INSERT: id[integer]:9
  COMMIT
-(27 rows)
+(28 rows)
 
 SELECT pg_drop_replication_slot('regression_slot');
  pg_drop_replication_slot 
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
@@ -570,8 +570,13 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	 * Process invalidation messages, even if we're not interested in the
 	 * transaction's contents, since the various caches need to always be
 	 * consistent.
+	 *
+	 * Also if that transaction was sent to prepare callback then both
+	 * this function were called during prepare.
 	 */
-	if (parsed->nmsgs > 0)
+	if (parsed->nmsgs > 0 &&
+			!(TransactionIdIsValid(parsed->twophase_xid) &&
+				ReorderBufferTxnIsPrepared(ctx->reorder, xid)))
 	{
 		ReorderBufferAddInvalidations(ctx->reorder, xid, buf->origptr,
 									  parsed->nmsgs, parsed->msgs);
@@ -634,8 +639,10 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		 * empty. So we can skip use shortcut for coomiting bare xact.
 		 */
 		ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
-							commit_time, origin_id, origin_lsn, parsed->twophase_gid, true);
-	} else {
+					commit_time, origin_id, origin_lsn, parsed->twophase_gid, true);
+	}
+	else
+	{
 		/* replay actions of all transaction + subtransactions in order */
 		ReorderBufferCommit(ctx->reorder, xid, buf->origptr, buf->endptr,
 							commit_time, origin_id, origin_lsn);
@@ -724,7 +731,9 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	 * If that is ROLLBACK PREPARED than send that to callbacks.
 	 */
 	if (TransactionIdIsValid(xid) &&
+			!SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) &&
 			parsed->dbId == ctx->slot->data.database &&
+			!FilterByOrigin(ctx, origin_id) &&
 			ReorderBufferTxnIsPrepared(ctx->reorder, xid))
 	{
 		ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
@@ -1633,8 +1633,12 @@ ReorderBufferCommitInternal(ReorderBufferTXN *txn,
 		if (snapshot_now->copied)
 			ReorderBufferFreeSnap(rb, snapshot_now);
 
-		/* remove potential on-disk data, and deallocate */
-		ReorderBufferCleanupTXN(rb, txn);
+		/*
+		 * remove potential on-disk data, and deallocate or postpone that
+		 * till the finish of two-phase tx
+		 */
+		if (!txn->prepared)
+			ReorderBufferCleanupTXN(rb, txn);
 	}
 	PG_CATCH();
 	{
@@ -1735,6 +1739,10 @@ ReorderBufferTxnIsPrepared(ReorderBuffer *rb, TransactionId xid)
 	txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
 								false);
 
+	/*
+	 * If txn == NULL then presumably subscriber confirmed prepare
+	 * but we are rebooted.
+	 */
 	return txn == NULL ? true : txn->prepared;
 }
 
@@ -1765,6 +1773,7 @@ ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,
 	else
 		rb->abort_prepared(rb, txn, commit_lsn);
 
+	ReorderBufferCleanupTXN(rb, txn);
 }
 
 /*
