Rergister custom xlog reader callbacks for on-demand WAL download in …

…StartupDecodingContext
neondatabase · Sep 16, 2024 · 221d2c1 · 221d2c1
1 parent 0baa734
commit 221d2c1
Show file tree

Hide file tree

Showing 7 changed files with 9 additions and 22 deletions.
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
@@ -43,6 +43,8 @@
 #include "utils/builtins.h"
 #include "utils/memutils.h"
 
+void		(*Custom_XLogReaderRoutines)(XLogReaderRoutine *xlr);
+
 /* data for errcontext callback */
 typedef struct LogicalErrorCallbackState
 {
@@ -180,6 +182,12 @@ StartupDecodingContext(List *output_plugin_options,
 	if (!fast_forward)
 		LoadOutputPlugin(&ctx->callbacks, NameStr(slot->data.plugin));
 
+	/*
+	 * NEON: override page_read/segment_open/segment_close functions to support on-demand WAL download
+	 */
+	if (Custom_XLogReaderRoutines != NULL)
+		Custom_XLogReaderRoutines(xl_routine);
+
 	/*
 	 * Now that the slot's xmin has been set, we can announce ourselves as a
 	 * logical decoding backend which doesn't need to be checked individually

diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
@@ -40,8 +40,6 @@
 #include "utils/regproc.h"
 #include "utils/resowner.h"
 
-void		(*LogicalFuncs_Custom_XLogReaderRoutines)(XLogReaderRoutine *xlr);
-
 /* Private data for writing out data */
 typedef struct DecodingOutputState
 {
@@ -211,9 +209,6 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
 		xlr.segment_open = wal_segment_open;
 		xlr.segment_close = wal_segment_close;
 
-		if (LogicalFuncs_Custom_XLogReaderRoutines != NULL)
-			LogicalFuncs_Custom_XLogReaderRoutines(&xlr);
-
 		/* restart at slot's confirmed_flush */
 		ctx = CreateDecodingContext(InvalidXLogRecPtr,
 									options,

diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
@@ -26,8 +26,6 @@
 #include "utils/pg_lsn.h"
 #include "utils/resowner.h"
 
-void		(*SlotFuncs_Custom_XLogReaderRoutines)(XLogReaderRoutine *xlr);
-
 /*
  * Helper function for creating a new physical replication slot with
  * given arguments. Note that this function doesn't release the created
@@ -128,9 +126,6 @@ create_logical_replication_slot(char *name, char *plugin,
 	xlr.segment_open = wal_segment_open;
 	xlr.segment_close = wal_segment_close;
 
-	if (SlotFuncs_Custom_XLogReaderRoutines != NULL)
-		SlotFuncs_Custom_XLogReaderRoutines(&xlr);
-
 	Assert(!MyReplicationSlot);
 
 	/*
@@ -490,9 +485,6 @@ pg_logical_replication_slot_advance(XLogRecPtr moveto)
 		xlr.segment_open = wal_segment_open;
 		xlr.segment_close = wal_segment_close;
 
-		if (SlotFuncs_Custom_XLogReaderRoutines != NULL)
-			SlotFuncs_Custom_XLogReaderRoutines(&xlr);
-
 		/*
 		 * Create our decoding context in fast_forward mode, passing start_lsn
 		 * as InvalidXLogRecPtr, so that we start processing from my slot's

diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
@@ -126,7 +126,6 @@ int			wal_sender_timeout = 60 * 1000; /* maximum time to send one WAL
 											 * data message */
 bool		log_replication_commands = false;
 
-void		(*WalSender_Custom_XLogReaderRoutines)(XLogReaderRoutine *xlr);
 /*
  * State for WalSndWakeupRequest
  */
@@ -1285,8 +1284,6 @@ StartLogicalReplication(StartReplicationCmd *cmd)
 	xlr.page_read = logical_read_xlog_page;
 	xlr.segment_open = WalSndSegmentOpen;
 	xlr.segment_close = wal_segment_close;
-	if (WalSender_Custom_XLogReaderRoutines != NULL)
-		WalSender_Custom_XLogReaderRoutines(&xlr);
 
 	/*
 	 * Create our decoding context, making it start at the previously ack'ed

diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
@@ -117,7 +117,7 @@ typedef struct LogicalDecodingContext
 	bool		in_create;
 } LogicalDecodingContext;
 
-extern void		(*LogicalFuncs_Custom_XLogReaderRoutines)(XLogReaderRoutine *xlr);
+extern void		(*Custom_XLogReaderRoutines)(XLogReaderRoutine *xlr);
 
 extern void CheckLogicalDecodingRequirements(void);
 

diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
@@ -183,8 +183,6 @@ typedef struct ReplicationSlot
 #define SlotIsPhysical(slot) ((slot)->data.database == InvalidOid)
 #define SlotIsLogical(slot) ((slot)->data.database != InvalidOid)
 
-extern void		(*SlotFuncs_Custom_XLogReaderRoutines)(XLogReaderRoutine *xlr);
-
 /*
  * Shared memory control area for all of replication slots.
  */

diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
@@ -36,9 +36,6 @@ extern PGDLLIMPORT int max_wal_senders;
 extern PGDLLIMPORT int wal_sender_timeout;
 extern PGDLLIMPORT bool log_replication_commands;
 
-struct XLogReaderRoutine;
-extern PGDLLIMPORT void (*WalSender_Custom_XLogReaderRoutines)(struct XLogReaderRoutine *xlr);
-
 extern void InitWalSender(void);
 extern bool exec_replication_command(const char *cmd_string);
 extern void WalSndErrorCleanup(void);