Properly set committed subtransactions in snapbuilder.

Author: kelvich

src/backend/replication/logical/decode.c

@@ -579,7 +579,7 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	}
 
 	SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
-					   parsed->nsubxacts, parsed->subxacts, true);
+					   parsed->nsubxacts, parsed->subxacts, true, parsed->nmsgs > 0);
 
 	/* ----
 	 * Check whether we are interested in this specific transaction, and tell
@@ -646,7 +646,7 @@ 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);
+						   parsed->nsubxacts, parsed->subxacts, true, false);
 		ReorderBufferCommitBareXact(ctx->reorder, xid, buf->origptr, buf->endptr,
 							commit_time, origin_id, origin_lsn);
 	} else {

src/backend/replication/logical/snapbuild.c

@@ -948,7 +948,7 @@ SnapBuildAbortTxn(SnapBuild *builder, XLogRecPtr lsn,
  */
 void
 SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
-				   int nsubxacts, TransactionId *subxacts, bool isCommit)
+				   int nsubxacts, TransactionId *subxacts, bool isCommit, bool forceCatalog)
 {
 	int			nxact;
 
@@ -1006,7 +1006,7 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
 		 * Add subtransaction to base snapshot if it DDL, we don't distinguish
 		 * to toplevel transactions there.
 		 */
-		else if (ReorderBufferXidHasCatalogChanges(builder->reorder, subxid))
+		else if (ReorderBufferXidHasCatalogChanges(builder->reorder, subxid) || forceCatalog)
 		{
 			sub_needs_timetravel = true;
 
@@ -1034,7 +1034,7 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
 		SnapBuildAddCommittedTxn(builder, xid);
 	}
 	/* add toplevel transaction to base snapshot */
-	else if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))
+	else if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid) || forceCatalog)
 	{
 		elog(DEBUG2, "found top level transaction " XID_FMT ", with catalog changes!",
 			 xid);

src/include/replication/snapbuild.h

@@ -70,7 +70,7 @@ extern bool SnapBuildXactNeedsSkip(SnapBuild *snapstate, XLogRecPtr ptr);
 
 extern void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn,
 				   TransactionId xid, int nsubxacts,
-				   TransactionId *subxacts, bool isCommit);
+				   TransactionId *subxacts, bool isCommit, bool forceCatalog);
 extern void SnapBuildAbortTxn(SnapBuild *builder, XLogRecPtr lsn,
 				  TransactionId xid, int nsubxacts,
 				  TransactionId *subxacts);