diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c
index ae5f397f35..de1b692658 100644
--- a/contrib/test_decoding/test_decoding.c
+++ b/contrib/test_decoding/test_decoding.c
@@ -77,6 +77,7 @@ static void pg_decode_message(LogicalDecodingContext *ctx,
bool transactional, const char *prefix,
Size sz, const char *message);
static bool pg_decode_filter_prepare(LogicalDecodingContext *ctx,
+ TransactionId xid,
const char *gid);
static void pg_decode_begin_prepare_txn(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn);
@@ -440,7 +441,8 @@ pg_decode_rollback_prepared_txn(LogicalDecodingContext *ctx,
* substring, then we filter it out.
*/
static bool
-pg_decode_filter_prepare(LogicalDecodingContext *ctx, const char *gid)
+pg_decode_filter_prepare(LogicalDecodingContext *ctx, TransactionId xid,
+ const char *gid)
{
if (strstr(gid, "_nodecode") != NULL)
return true;
diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index 80eb96d609..da23f89ca3 100644
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -794,20 +794,25 @@ typedef void (*LogicalDecodeMessageCB) (struct LogicalDecodingContext *ctx,
COMMIT PREPARED time. To signal that
decoding should be skipped, return true;
false otherwise. When the callback is not
- defined, false is assumed (i.e. nothing is
- filtered).
+ defined, false is assumed (i.e. no filtering, all
+ transactions using two-phase commit are decoded in two phases as well).
typedef bool (*LogicalDecodeFilterPrepareCB) (struct LogicalDecodingContext *ctx,
+ TransactionId xid,
const char *gid);
- The ctx parameter has the same contents as for the
- other callbacks. The gid is the identifier that later
- identifies this transaction for COMMIT PREPARED or
- ROLLBACK PREPARED.
+ The ctx parameter has the same contents as for
+ the other callbacks. The parameters xid
+ and gid provide two different ways to identify
+ the transaction. The later COMMIT PREPARED or
+ ROLLBACK PREPARED carries both identifiers,
+ providing an output plugin the choice of what to use.
- The callback has to provide the same static answer for a given
- gid every time it is called.
+ The callback may be invoked multiple times per transaction to decode
+ and must provide the same static answer for a given pair of
+ xid and gid every time
+ it is called.
@@ -1219,9 +1224,11 @@ stream_commit_cb(...); <-- commit of the streamed transaction
- Optionally the output plugin can specify a name pattern in the
- filter_prepare_cb and transactions with gid containing
- that name pattern will not be decoded as a two-phase commit transaction.
+ Optionally the output plugin can define filtering rules via
+ filter_prepare_cb to decode only specific transaction
+ in two phases. This can be achieved by pattern matching on the
+ gid or via lookups using the
+ xid.
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index 5f596135b1..97be4b0f23 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 37b75deb72..2f6803637b 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;
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index c253403372..af551d6f4e 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -125,7 +125,8 @@ extern void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn,
XLogRecPtr restart_lsn);
extern void LogicalConfirmReceivedLocation(XLogRecPtr lsn);
-extern bool filter_prepare_cb_wrapper(LogicalDecodingContext *ctx, const char *gid);
+extern bool filter_prepare_cb_wrapper(LogicalDecodingContext *ctx,
+ TransactionId xid, const char *gid);
extern bool filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id);
extern void ResetLogicalStreamingState(void);
extern void UpdateDecodingStats(LogicalDecodingContext *ctx);
diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h
index 2c2c964c55..810495ed0e 100644
--- a/src/include/replication/output_plugin.h
+++ b/src/include/replication/output_plugin.h
@@ -106,6 +106,7 @@ typedef void (*LogicalDecodeShutdownCB) (struct LogicalDecodingContext *ctx);
* and sent as usual transaction.
*/
typedef bool (*LogicalDecodeFilterPrepareCB) (struct LogicalDecodingContext *ctx,
+ TransactionId xid,
const char *gid);
/*