clean snapshot

kelvich · kelvich · commit 91893538384b · 2017-03-27T17:06:25.000+03:00
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
@@ -663,7 +663,7 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 		ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr);
 	}
 
-	SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
+	SnapBuildPrepareTxnStart(ctx->snapshot_builder, buf->origptr, xid,
 					   parsed->nsubxacts, parsed->subxacts);
 
 	if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
@@ -689,6 +689,8 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	/* replay actions of all transaction + subtransactions in order */
 	ReorderBufferPrepare(ctx->reorder, xid, buf->origptr, buf->endptr,
 						commit_time, origin_id, origin_lsn, parsed->twophase_gid);
+
+	SnapBuildPrepareTxnFinish(ctx->snapshot_builder, xid);
 }
 
 /*
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
@@ -868,7 +868,7 @@ SnapBuildPurgeCommittedTxn(SnapBuild *builder)
 	/* copy xids that still are interesting to workspace */
 	for (off = 0; off < builder->committed.xcnt; off++)
 	{
-		if (NormalTransactionIdPrecedes(builder->committed.xip[off],
+		if (TransactionIdPrecedes(builder->committed.xip[off],
 										builder->xmin))
 			;					/* remove */
 		else
@@ -1101,6 +1101,29 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
 	}
 }
 
+void
+SnapBuildPrepareTxnStart(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
+				   int nsubxacts, TransactionId *subxacts)
+{
+	SnapBuildCommitTxn(builder, lsn, xid, nsubxacts, subxacts);
+}
+
+void
+SnapBuildPrepareTxnFinish(SnapBuild *builder, TransactionId xid)
+{
+	TransactionId *search = bsearch(&xid, builder->running.xip,
+				builder->running.xcnt_space, sizeof(TransactionId), xidComparator);
+
+	if (search == NULL)
+		return;
+
+	/* delete that xid */
+	memmove(search, search + 1,
+			((builder->running.xip + builder->running.xcnt - 1) - search) * sizeof(TransactionId));
+	builder->running.xcnt--;
+	builder->running.xmin = builder->running.xip[0];
+	builder->running.xmax = builder->running.xip[builder->running.xcnt - 1];
+}
 
 /* -----------------------------------
  * Snapshot building functions dealing with xlog records
diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h
@@ -71,6 +71,10 @@ extern bool SnapBuildXactNeedsSkip(SnapBuild *snapstate, XLogRecPtr ptr);
 extern void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn,
 				   TransactionId xid, int nsubxacts,
 				   TransactionId *subxacts);
+extern void SnapBuildPrepareTxnStart(SnapBuild *builder, XLogRecPtr lsn,
+				   TransactionId xid, int nsubxacts,
+				   TransactionId *subxacts);
+extern void SnapBuildPrepareTxnFinish(SnapBuild *builder, TransactionId xid);
 extern void SnapBuildAbortTxn(SnapBuild *builder, XLogRecPtr lsn,
 				  TransactionId xid, int nsubxacts,
 				  TransactionId *subxacts);

Original file line number	Diff line number	Diff line change
`@@ -663,7 +663,7 @@ DecodePrepare(LogicalDecodingContext ctx, XLogRecordBuffer buf,`
`663`	`663`	`ReorderBufferXidSetCatalogChanges(ctx->reorder, xid, buf->origptr);`
`664`	`664`	`}`
`665`	`665`
`666`		`- SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,`
	`666`	`+ SnapBuildPrepareTxnStart(ctx->snapshot_builder, buf->origptr, xid,`
`667`	`667`	`parsed->nsubxacts, parsed->subxacts);`
`668`	`668`
`669`	`669`	`if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) \|\|`
`@@ -689,6 +689,8 @@ DecodePrepare(LogicalDecodingContext ctx, XLogRecordBuffer buf,`
`689`	`689`	`/* replay actions of all transaction + subtransactions in order */`
`690`	`690`	`ReorderBufferPrepare(ctx->reorder, xid, buf->origptr, buf->endptr,`
`691`	`691`	`commit_time, origin_id, origin_lsn, parsed->twophase_gid);`
	`692`	`+`
	`693`	`+ SnapBuildPrepareTxnFinish(ctx->snapshot_builder, xid);`
`692`	`694`	`}`
`693`	`695`
`694`	`696`	`/*`