diff options
Diffstat (limited to 'src/backend/replication/logical')
| -rw-r--r-- | src/backend/replication/logical/decode.c | 17 | ||||
| -rw-r--r-- | src/backend/replication/logical/logical.c | 5 |
2 files changed, 14 insertions, 8 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 diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 37b75deb728..2f6803637bf 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -1083,7 +1083,8 @@ truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn, } bool -filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, const char *gid) +filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, TransactionId xid, + const char *gid) { LogicalErrorCallbackState state; ErrorContextCallback errcallback; @@ -1104,7 +1105,7 @@ filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, const char *gid) ctx->accept_writes = false; /* do the actual work: call callback */ - ret = ctx->callbacks.filter_prepare_cb(ctx, gid); + ret = ctx->callbacks.filter_prepare_cb(ctx, xid, gid); /* Pop the error context stack */ error_context_stack = errcallback.previous; |