Move SnapBuildCommitTxn() from prepare to commit/abort.

Previous behaviour was wrong as in case of abort transaction
were vesible in snapshot.

Author: kelvich

src/backend/replication/logical/decode.c

@@ -645,6 +645,8 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		 */
 		strcpy(ctx->reorder->gid, parsed->twophase_gid);
 		*ctx->reorder->state_3pc = '\0';
+		SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
+						   parsed->nsubxacts, parsed->subxacts, true);
 		ReorderBufferCommitBareXact(ctx->reorder, xid, buf->origptr, buf->endptr,
 							commit_time, origin_id, origin_lsn);
 	} else {
@@ -699,9 +701,6 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		ReorderBufferXidSetCatalogChanges(rb, xid, buf->origptr);
 	}
 
-	SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
-					   parsed->nsubxacts, parsed->subxacts, false);
-
 	if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
 		(parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||
 		FilterByOrigin(ctx, origin_id))
@@ -777,6 +776,8 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 			strcpy(ctx->reorder->gid, parsed->twophase_gid);
 			*ctx->reorder->state_3pc = '\0';
 
+			SnapBuildAbortTxn(ctx->snapshot_builder, buf->record->EndRecPtr, xid,
+							  parsed->nsubxacts, parsed->subxacts);
 			ReorderBufferCommitBareXact(ctx->reorder, xid, buf->origptr, buf->endptr,
 										commit_time, origin_id, origin_lsn);
 		}

src/backend/replication/logical/reorderbuffer.c

@@ -1728,6 +1728,10 @@ ReorderBufferCommitBareXact(ReorderBuffer *rb, TransactionId xid,
 	txn = ReorderBufferTXNByXid(rb, xid, true, NULL, commit_lsn,
 								true);
 
+	if (rb->xact_action == XLOG_XACT_ABORT_PREPARED) {
+		ReorderBufferCleanupTXN(rb, txn);
+	}
+
 	txn->final_lsn = commit_lsn;
 	txn->end_lsn = end_lsn;
 	txn->commit_time = commit_time;