diff options
Diffstat (limited to 'src/backend/replication/logical/decode.c')
| -rw-r--r-- | src/backend/replication/logical/decode.c | 17 |
1 files changed, 11 insertions, 6 deletions
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c index 5f596135b15..97be4b0f23f 100644 --- a/src/backend/replication/logical/decode.c +++ b/src/backend/replication/logical/decode.c @@ -80,7 +80,8 @@ static void DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tup); /* helper functions for decoding transactions */ -static inline bool FilterPrepare(LogicalDecodingContext *ctx, const char *gid); +static inline bool FilterPrepare(LogicalDecodingContext *ctx, + TransactionId xid, const char *gid); static bool DecodeTXNNeedSkip(LogicalDecodingContext *ctx, XLogRecordBuffer *buf, Oid dbId, RepOriginId origin_id); @@ -271,7 +272,8 @@ DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) * doesn't filter the transaction at prepare time. */ if (info == XLOG_XACT_COMMIT_PREPARED) - two_phase = !(FilterPrepare(ctx, parsed.twophase_gid)); + two_phase = !(FilterPrepare(ctx, xid, + parsed.twophase_gid)); DecodeCommit(ctx, buf, &parsed, xid, two_phase); break; @@ -298,7 +300,8 @@ DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) * doesn't filter the transaction at prepare time. */ if (info == XLOG_XACT_ABORT_PREPARED) - two_phase = !(FilterPrepare(ctx, parsed.twophase_gid)); + two_phase = !(FilterPrepare(ctx, xid, + parsed.twophase_gid)); DecodeAbort(ctx, buf, &parsed, xid, two_phase); break; @@ -355,7 +358,8 @@ DecodeXactOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) * manner iff output plugin supports two-phase commits and * doesn't filter the transaction at prepare time. */ - if (FilterPrepare(ctx, parsed.twophase_gid)) + if (FilterPrepare(ctx, parsed.twophase_xid, + parsed.twophase_gid)) { ReorderBufferProcessXid(reorder, parsed.twophase_xid, buf->origptr); @@ -581,7 +585,8 @@ DecodeHeapOp(LogicalDecodingContext *ctx, XLogRecordBuffer *buf) * this transaction as a regular commit later. */ static inline bool -FilterPrepare(LogicalDecodingContext *ctx, const char *gid) +FilterPrepare(LogicalDecodingContext *ctx, TransactionId xid, + const char *gid) { /* * Skip if decoding of two-phase transactions at PREPARE time is not @@ -599,7 +604,7 @@ FilterPrepare(LogicalDecodingContext *ctx, const char *gid) if (ctx->callbacks.filter_prepare_cb == NULL) return false; - return filter_prepare_cb_wrapper(ctx, gid); + return filter_prepare_cb_wrapper(ctx, xid, gid); } static inline bool |