Commit63c917c

committed

[PGPRO-4074] Port LogicalDecodingCaughtUp callback.

tags: multimaster(cherry picked from commit d75ea7f66dfea0644d0bf98011f4bc2fd32bcd63)

1 parent6a640de commit63c917cCopy full SHA for 63c917c

File tree

+47

-0

lines changed

+47

-0

lines changed

Lines changed: 29 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -1040,6 +1040,35 @@ message_cb_wrapper(ReorderBuffer cache, ReorderBufferTXN txn,`
`1040`	`1040`	`error_context_stack=errcallback.previous;`
`1041`	`1041`	`}`
`1042`	`1042`
	`1043`	`+voidLogicalDecodingCaughtUp(LogicalDecodingContext*ctx)`
	`1044`	`+{`
	`1045`	`+LogicalErrorCallbackStatestate;`
	`1046`	`+ErrorContextCallbackerrcallback;`
	`1047`	`+`
	`1048`	`+if (ctx->callbacks.caughtup_cb==NULL)`
	`1049`	`+return;`
	`1050`	`+`
	`1051`	`+/* Push callback + info on the error context stack */`
	`1052`	`+state.ctx=ctx;`
	`1053`	`+state.callback_name="caughtup";`
	`1054`	`+state.report_location=ctx->reader->EndRecPtr;`
	`1055`	`+errcallback.callback=output_plugin_error_callback;`
	`1056`	`+errcallback.arg= (void*)&state;`
	`1057`	`+errcallback.previous=error_context_stack;`
	`1058`	`+error_context_stack=&errcallback;`
	`1059`	`+`
	`1060`	`+/* set output state */`
	`1061`	`+ctx->accept_writes= true;`
	`1062`	`+ctx->write_xid=InvalidTransactionId;`
	`1063`	`+ctx->write_location=ctx->reader->EndRecPtr;`
	`1064`	`+`
	`1065`	`+/* do the actual work: call callback */`
	`1066`	`+ctx->callbacks.caughtup_cb(ctx);`
	`1067`	`+`
	`1068`	`+/* Pop the error context stack */`
	`1069`	`+error_context_stack=errcallback.previous;`
	`1070`	`+}`
	`1071`	`+`
`1043`	`1072`	`/*`
`1044`	`1073`	`* Set the required catalog xmin horizon for historic snapshots in the current`
`1045`	`1074`	`* replication slot.`

Lines changed: 10 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -1380,6 +1380,7 @@ WalSndWaitForWal(XLogRecPtr loc)`
`1380`	`1380`	`{`
`1381`	`1381`	`intwakeEvents;`
`1382`	`1382`	`staticXLogRecPtrRecentFlushPtr=InvalidXLogRecPtr;`
	`1383`	`+boolcaughtup_cb_called= false;`
`1383`	`1384`
`1384`	`1385`	`/*`
`1385`	`1386`	`* Fast path to avoid acquiring the spinlock in case we already know we`
`@@ -1459,6 +1460,15 @@ WalSndWaitForWal(XLogRecPtr loc)`
`1459`	`1460`
`1460`	`1461`	`/* Waiting for new WAL. Since we need to wait, we're now caught up. */`
`1461`	`1462`	`WalSndCaughtUp= true;`
	`1463`	`+/*`
	`1464`	`+ * Call cb only once: if it writes anyting, it'll probably call`
	`1465`	`+ * WalSndWriteData who sets latch, thus creating busy loop.`
	`1466`	`+ */`
	`1467`	`+if (!caughtup_cb_called&&logical_decoding_ctx)`
	`1468`	`+{`
	`1469`	`+LogicalDecodingCaughtUp(logical_decoding_ctx);`
	`1470`	`+caughtup_cb_called= true;`
	`1471`	`+}`
`1462`	`1472`
`1463`	`1473`	`/*`
`1464`	`1474`	`* Try to flush any pending output to the client.`

Lines changed: 2 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -120,6 +120,8 @@ extern void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn,`
`120`	`120`	`XLogRecPtrrestart_lsn);`
`121`	`121`	`externvoidLogicalConfirmReceivedLocation(XLogRecPtrlsn);`
`122`	`122`
	`123`	`+externvoidLogicalDecodingCaughtUp(LogicalDecodingContext*ctx);`
	`124`	`+`
`123`	`125`	`externboolfilter_by_origin_cb_wrapper(LogicalDecodingContext*ctx,RepOriginIdorigin_id);`
`124`	`126`
`125`	`127`	`#endif`

Lines changed: 6 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -139,6 +139,11 @@ typedef bool (LogicalDecodeFilterByOriginCB) (struct LogicalDecodingContext ct`
`139`	`139`	`*/`
`140`	`140`	`typedefvoid (LogicalDecodeShutdownCB) (structLogicalDecodingContextctx);`
`141`	`141`
	`142`	`+/*`
	`143`	`+ * Called when WAL sender caught up.`
	`144`	`+ */`
	`145`	`+typedefvoid (LogicalDecodeCaughtUpCB) (structLogicalDecodingContextctx);`
	`146`	`+`
`142`	`147`	`/*`
`143`	`148`	`* Output plugin callbacks`
`144`	`149`	`*/`
`@@ -157,6 +162,7 @@ typedef struct OutputPluginCallbacks`
`157`	`162`	`LogicalDecodeAbortPreparedCBabort_prepared_cb;`
`158`	`163`	`LogicalDecodeFilterByOriginCBfilter_by_origin_cb;`
`159`	`164`	`LogicalDecodeShutdownCBshutdown_cb;`
	`165`	`+LogicalDecodeCaughtUpCBcaughtup_cb;`
`160`	`166`	`}OutputPluginCallbacks;`
`161`	`167`
`162`	`168`	`/* Functions in replication/logical/logical.c */`

Comments

(0)