On Thu, Jun 21, 2018 at 07:31:20PM +0900, Michael Paquier wrote: > Could it be possible to get a patch from all the feedback and exchange > gathered here? Petr, I think that it would not hurt if you use the set > of words and comments you think is most adapted as the primary author of > the feature.
I have seen no patch, so attached is one to finally close the loop and this open item, which includes both my suggestions and what Arseny has mentioned based on the latest emails exchanged. Any objections to that? -- Michael
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index c2d0e0c723..7e10a027ca 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -340,6 +340,9 @@ CreateInitDecodingContext(char *plugin,
* output_plugin_options
* contains options passed to the output plugin.
*
+ * fast_forward
+ * bypasses the generation of logical changes.
+ *
* read_page, prepare_write, do_write, update_progress
* callbacks that have to be filled to perform the use-case dependent,
* actual work.
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 2806e1076c..6c2addd5b7 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -343,10 +343,10 @@ pg_physical_replication_slot_advance(XLogRecPtr moveto)
* Helper function for advancing logical replication slot forward.
* The slot's restart_lsn is used as start point for reading records,
* while confirmed_lsn is used as base point for the decoding context.
- * The LSN position to move to is checked by doing a per-record scan and
- * logical decoding which makes sure that confirmed_lsn is updated to a
- * LSN which allows the future slot consumer to get consistent logical
- * changes.
+ * While we could just use LogicalConfirmReceivedLocation to update
+ * confirmed_flush_lsn, we had better digest WAL to advance restart_lsn
+ * allowing to recycle WAL and old catalog tuples. As decoding is done
+ * with fast_forward mode, no changes are generated.
*/
static XLogRecPtr
pg_logical_replication_slot_advance(XLogRecPtr moveto)
@@ -358,7 +358,10 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
PG_TRY();
{
- /* restart at slot's confirmed_flush */
+ /*
+ * Note that start_lsn does not matter here, as with fast_forward mode
+ * no transactions are replayed.
+ */
ctx = CreateDecodingContext(InvalidXLogRecPtr,
NIL,
true,
@@ -378,6 +381,10 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
XLogRecord *record;
char *errm = NULL;
+ /*
+ * Start reading WAL at the slot's restart_lsn, which certainly
+ * points to the valid record.
+ */
record = XLogReadRecord(ctx->reader, startlsn, &errm);
if (errm)
elog(ERROR, "%s", errm);
@@ -389,8 +396,8 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
startlsn = InvalidXLogRecPtr;
/*
- * The {begin_txn,change,commit_txn}_wrapper callbacks above will
- * store the description into our tuplestore.
+ * Note that changes are not generated in fast_forward mode, and
+ * that the slot's data is still updated.
*/
if (record != NULL)
LogicalDecodingProcessRecord(ctx, ctx->reader);
signature.asc
Description: PGP signature
