Offer to do 2PC only if snap is consistent.

arssher · arssher · commit f46f9d8248fa · 2018-09-20T10:57:59.000+03:00
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
@@ -287,8 +287,13 @@ DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 			{
 				xl_xact_parsed_prepare parsed;
 
-				/* check that output plugin is capable of twophase decoding */
-				if (!ctx->options.enable_twophase)
+				/*
+				 * Check that output plugin is capable of twophase decoding.
+				 * We also don't offer to do 2PC if snap is not yet consistent
+				 * as of reading PREPARE.
+				 */
+				if (!ctx->options.enable_twophase ||
+					SnapBuildCurrentState(builder) < SNAPBUILD_CONSISTENT)
 				{
 					ReorderBufferProcessXid(reorder, XLogRecGetXid(r), buf->origptr);
 					break;
@@ -584,13 +589,23 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	TimestampTz commit_time = parsed->xact_time;
 	RepOriginId origin_id = XLogRecGetOrigin(buf->record);
 	int			i;
+	bool		reorderbuffer_has_xid;
 
 	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
 	{
 		origin_lsn = parsed->origin_lsn;
 		commit_time = parsed->origin_timestamp;
 	}
 
+	/*
+	 * If this is COMMIT PREPARED and ReorderBuffer doesn't have this xid,
+	 * either the plugin refused to do 2PC on this xact or we didn't have
+	 * consistent snapshot yet during PREPARE processing. Anyway, in this case
+	 * we don't do 2PC and replay xact fully now. We must check this early
+	 * since invalidation addition below might add the record to the RB.
+	 */
+	reorderbuffer_has_xid = ReorderBufferHasXid(ctx->reorder, xid);
+
 	/*
 	 * Process invalidation messages, even if we're not interested in the
 	 * transaction's contents, since the various caches need to always be
@@ -663,10 +678,14 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	 * For output plugins that do not support PREPARE-time decoding of
 	 * two-phase transactions, we never even see the PREPARE and all two-phase
 	 * transactions simply fall through to the second branch.
+	 *
+	 * We rely on existence of xid in reorderbuffer to determine was 2PC done
+	 * or not. This is correct because we always see PREPARE before COMMIT
+	 * PREPARED if the latter was after consistent point.
+	 *
 	 */
 	if (TransactionIdIsValid(parsed->twophase_xid) &&
-		ReorderBufferTxnIsPrepared(ctx->reorder,
-								   parsed->twophase_xid, parsed->twophase_gid))
+		!reorderbuffer_has_xid)
 	{
 		Assert(xid == parsed->twophase_xid);
 		/* we are processing COMMIT PREPARED */
@@ -765,7 +784,7 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		!SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) &&
 		parsed->dbId == ctx->slot->data.database &&
 		!FilterByOrigin(ctx, origin_id) &&
-		ReorderBufferTxnIsPrepared(ctx->reorder, xid, parsed->twophase_gid))
+		!ReorderBufferHasXid(ctx->reorder, xid))
 	{
 		ReorderBufferFinishPrepared(ctx->reorder, xid, buf->origptr, buf->endptr,
 									commit_time, origin_id, origin_lsn,
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
@@ -1932,27 +1932,6 @@ ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
 								commit_time, origin_id, origin_lsn);
 }
 
-/*
- * Check whether this transaction was sent as prepared to subscribers.
- * Called while handling commit|abort prepared.
- */
-bool
-ReorderBufferTxnIsPrepared(ReorderBuffer *rb, TransactionId xid,
-						   const char *gid)
-{
-	ReorderBufferTXN *txn;
-
-	txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
-								false);
-
-	/*
-	 * Always call the prepare filter. It's the job of the prepare filter to
-	 * give us the *same* response for a given xid across multiple calls
-	 * (including ones on restart)
-	 */
-	return !(rb->filter_prepare(rb, txn, xid, gid));
-}
-
 /*
  * Send standalone xact event. This is used to handle COMMIT/ABORT PREPARED.
  */
@@ -2175,6 +2154,18 @@ ReorderBufferProcessXid(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
 		ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
 }
 
+/*
+ * Does reorderbuffer currently holds this xact?
+ */
+bool
+ReorderBufferHasXid(ReorderBuffer *rb, TransactionId xid)
+{
+	ReorderBufferTXN *txn;
+
+	txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, true);
+	return txn != NULL;
+}
+
 /*
  * Add a new snapshot to this transaction that may only used after lsn 'lsn'
  * because the previous snapshot doesn't describe the catalog correctly for
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
@@ -497,14 +497,13 @@ void ReorderBufferAddInvalidations(ReorderBuffer *, TransactionId, XLogRecPtr ls
 void ReorderBufferImmediateInvalidation(ReorderBuffer *, uint32 ninvalidations,
 								   SharedInvalidationMessage *invalidations);
 void		ReorderBufferProcessXid(ReorderBuffer *, TransactionId xid, XLogRecPtr lsn);
+bool		ReorderBufferHasXid(ReorderBuffer *, TransactionId xid);
 void		ReorderBufferXidSetCatalogChanges(ReorderBuffer *, TransactionId xid, XLogRecPtr lsn);
 bool		ReorderBufferXidHasCatalogChanges(ReorderBuffer *, TransactionId xid);
 bool		ReorderBufferXidHasBaseSnapshot(ReorderBuffer *, TransactionId xid);
 
 bool ReorderBufferPrepareNeedSkip(ReorderBuffer *rb, TransactionId xid,
 							 const char *gid);
-bool ReorderBufferTxnIsPrepared(ReorderBuffer *rb, TransactionId xid,
-						   const char *gid);
 void ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
 					 XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
 					 TimestampTz commit_time,
