some comments

Author: kelvich

contrib/test_decoding/test_decoding.c

@@ -279,32 +279,48 @@ pg_decode_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 	OutputPluginWrite(ctx, true);
 }
 
+
+/* Filter out unnecessary two-phase transactions */
 static bool
 pg_filter_prepare(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 					char *gid)
 {
 	TestDecodingData *data = ctx->output_plugin_private;
 
+	/* treat all transaction as one-phase */
 	if (!data->twophase_decoding)
 		return true;
 
+	/*
+	 * Two-phase transactions that accessed catalog require special treatment.
+	 *
+	 * Right now we don't have a save way to decode catalog changes made in
+	 * prepared transaction that was already aborted by the time of decoding.
+	 *
+	 * That kind of problem arises only when we are trying to retrospectively
+	 * decode aborted transactions. If one wants to code distributed commit
+	 * based on prepare decoding then commits/aborts will happend strictly after
+	 * decoding will be completed, so it is safe to skip any checks/locks here.
+	 */
 	if (txn->has_catalog_changes)
 	{
 		LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
 
 		if (TransactionIdIsInProgress(txn->xid))
 		{
 			/*
-			 * XXX
+			 * For the sake of simplicity we just ignore in-progess transaction
+			 * in this extension, as they may abort during deconing.
 			 */
 			LWLockRelease(TwoPhaseStateLock);
 			return true;
 		}
 		else if (TransactionIdDidAbort(txn->xid))
 		{
 			/*
-			 * Here we know that it is already aborted and should humble
-			 * ourselves.
+			 * Here we know that it is already aborted and there is no
+			 * mush sence in doing something with this transaction.
+			 * Consequent ABORT PREPARED will be suppressed.
 			 */
 			LWLockRelease(TwoPhaseStateLock);
 			return true;

src/backend/replication/logical/decode.c

@@ -295,7 +295,8 @@ DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf)
 										XLogRecGetData(buf->record), &parsed);
 
 				/* does output plugin wants this particular transaction? */
-				if (ReorderBufferPrepareNeedSkip(reorder, parsed.twophase_xid,
+				if (ctx->callbacks.filter_prepare_cb &&
+					ReorderBufferPrepareNeedSkip(reorder, parsed.twophase_xid,
 													 parsed.twophase_gid))
 				{
 					ReorderBufferProcessXid(reorder, parsed.twophase_xid,
@@ -641,22 +642,23 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	}
 }
 
+
+/*
+ * Decode PREPARE record. Same logic as in COMMIT, but diffent calls
+ * to SnapshotBuilder as we need to mark this transaction as commited
+ * instead of running to properly decode it. When prepared transation
+ * is decoded we mark it in snapshot as running again.
+ */
 static void
 DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 			 xl_xact_parsed_prepare *parsed)
 {
-	XLogRecPtr	origin_lsn = InvalidXLogRecPtr;
-	TimestampTz	commit_time = 0;
+	XLogRecPtr	origin_lsn = parsed->origin_lsn;
+	TimestampTz	commit_time = parsed->origin_timestamp;
 	XLogRecPtr	origin_id = XLogRecGetOrigin(buf->record);
 	int			i;
 	TransactionId xid = parsed->twophase_xid;
 
-	if (parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
-	{
-		origin_lsn = parsed->origin_lsn;
-		commit_time = parsed->origin_timestamp;
-	}
-
 	/*
 	 * Process invalidation messages, even if we're not interested in the
 	 * transaction's contents, since the various caches need to always be

src/backend/replication/logical/logical.c

@@ -815,7 +815,7 @@ filter_prepare_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, char *gid
 	error_context_stack = &errcallback;
 
 	/* set output state */
-	ctx->accept_writes = true; // ->false
+	ctx->accept_writes = false;
 
 	/* do the actual work: call callback */
 	ret = ctx->callbacks.filter_prepare_cb(ctx, txn, gid);

src/backend/replication/logical/reorderbuffer.c

@@ -1667,17 +1667,24 @@ ReorderBufferCommitInternal(ReorderBufferTXN *txn,
 	PG_END_TRY();
 }
 
+/*
+ * Ask output plugin whether we want to skip this PREPARE and send
+ * this transaction as one-phase later on commit.
+ */
 bool
 ReorderBufferPrepareNeedSkip(ReorderBuffer *rb, TransactionId xid, char *gid)
 {
 	ReorderBufferTXN *txn;
 
 	txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, false);
-	Assert(txn != NULL);
-	// ctx->callbacks.filter_prepare_cb
+
 	return rb->filter_prepare(rb, txn, gid);
 }
 
+
+/*
+ * Commit non-twophase transaction. See comments to ReorderBufferCommitInternal()
+ */
 void
 ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
 					XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
@@ -1693,6 +1700,10 @@ ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
 										commit_time, origin_id, origin_lsn);
 }
 
+/*
+ * Prepare twophase transaction. It calls ReorderBufferCommitInternal()
+ * since all transaction changes should be decoded on PREPARE.
+ */
 void
 ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
 					XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
@@ -1712,6 +1723,10 @@ ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid,
 										commit_time, origin_id, origin_lsn);
 }
 
+/*
+ * Check whether this transaction was sent as prepared to receiver.
+ * Called upon commit/abort prepared.
+ */
 bool
 ReorderBufferTxnIsPrepared(ReorderBuffer *rb, TransactionId xid)
 {
@@ -1720,7 +1735,6 @@ ReorderBufferTxnIsPrepared(ReorderBuffer *rb, TransactionId xid)
 	txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr,
 								false);
 
-	/* TXN not found mean that it was send already, isn't it? */
 	return txn == NULL ? true : txn->prepared;
 }
 

src/backend/replication/logical/snapbuild.c

@@ -1101,18 +1101,39 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
 	}
 }
 
+/*
+ * Just a wrapper to clarify DecodePrepare().
+ * Right now we can't extract correct historic catalog data that
+ * was produced by aborted prepared transaction, so it work of
+ * decoding plugin to avoid such situation and here we just construct usual
+ * snapshot to able to decode prepare.
+ */
 void
 SnapBuildPrepareTxnStart(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
 				   int nsubxacts, TransactionId *subxacts)
 {
 	SnapBuildCommitTxn(builder, lsn, xid, nsubxacts, subxacts);
 }
 
+
+/*
+ * When decoding of preppare is finished we want should exclude our xid
+ * from list of committed xids to have correct snapshot between prepare
+ * and commit.
+ *
+ * However, this is not sctrictly needed. Prepared transaction holds locks
+ * between prepare and commit so nodody can produce new version of our
+ * catalog tuples. In case of abort we will have this xid in array of
+ * commited xids, but it also will not cause a problem since checks of
+ * HeapTupleHeaderXminInvalid() in HeapTupleSatisfiesHistoricMVCC()
+ * have higher priority then checks for xip array. Anyway let's be consistent
+ * about definitions and delete this xid from xip array.
+ */
 void
 SnapBuildPrepareTxnFinish(SnapBuild *builder, TransactionId xid)
 {
 	TransactionId *search = bsearch(&xid, builder->running.xip,
-				builder->running.xcnt_space, sizeof(TransactionId), xidComparator);
+				builder->running.xcnt, sizeof(TransactionId), xidComparator);
 
 	if (search == NULL)
 		return;
@@ -1121,6 +1142,8 @@ SnapBuildPrepareTxnFinish(SnapBuild *builder, TransactionId xid)
 	memmove(search, search + 1,
 			((builder->running.xip + builder->running.xcnt - 1) - search) * sizeof(TransactionId));
 	builder->running.xcnt--;
+
+	/* update min/max */
 	builder->running.xmin = builder->running.xip[0];
 	builder->running.xmax = builder->running.xip[builder->running.xcnt - 1];
 }

src/include/replication/output_plugin.h

@@ -71,7 +71,7 @@ typedef void (*LogicalDecodeCommitCB) (struct LogicalDecodingContext *ctx,
  * Called before decoding of PREPARE record to decide whether this
  * transaction should be decoded with separate calls to prepare
  * and commit_prepared/abort_prepared callbacks or wait till COMMIT PREPARED
- * and send as usual transaction.
+ * and sent as usual transaction.
  */
 typedef bool (*LogicalDecodeFilterPrepareCB) (struct LogicalDecodingContext *ctx,
 													  ReorderBufferTXN *txn,

src/include/replication/reorderbuffer.h

@@ -149,9 +149,9 @@ typedef struct ReorderBufferTXN
 	char		gid[GIDSIZE];
 
 	/*
-	 * We have ability to treat twophase transaction as ordinary one
-	 * with the help of filter_prepare callback.
-	 * XXX: try to reword that comment
+	 * By using filter_prepare() callback we can force decoding to treat
+	 * two-phase transaction as on ordinary one. This flag is set if we are
+	 * actually called prepape() callback in output plugin.
 	 */
 	bool		prepared;