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

…tartupDecodingContext (#9007)

## Problem

See #8931
On-demand WAL download are not set in all cases where WAL is accessed by
logical replication

## Summary of changes

Set customer xlog reader handles in StartupDecodingContext

Related changes in Postgres modules:

neondatabase/postgres#495
neondatabase/postgres#496
neondatabase/postgres#497
neondatabase/postgres#498

## Checklist before requesting a review

- [ ] I have performed a self-review of my code.
- [ ] If it is a core feature, I have added thorough tests.
- [ ] Do we need to implement analytics? if so did you add the relevant
metrics to the dashboard?
- [ ] If this PR requires public announcement, mark it with
/release-notes label and add several sentences in this section.

## Checklist before merging

- [ ] Do not forget to reformat commit message to not include the above
checklist

---------

Co-authored-by: Konstantin Knizhnik <knizhnik@neon.tech>

pgxn/neon/neon.c

@@ -421,9 +421,7 @@ _PG_init(void)
 
 	pg_init_libpagestore();
 	pg_init_walproposer();
-	WalSender_Custom_XLogReaderRoutines = NeonOnDemandXLogReaderRoutines;
-	LogicalFuncs_Custom_XLogReaderRoutines = NeonOnDemandXLogReaderRoutines;
-	SlotFuncs_Custom_XLogReaderRoutines = NeonOnDemandXLogReaderRoutines;
+	Custom_XLogReaderRoutines = NeonOnDemandXLogReaderRoutines;
 
 	InitUnstableExtensionsSupport();
 	InitLogicalReplicationMonitor();

test_runner/regress/test_ondemand_wal_download.py

@@ -0,0 +1,27 @@
+from fixtures.neon_fixtures import NeonEnv
+
+
+def test_on_demand_wal_download(neon_simple_env: NeonEnv):
+    env = neon_simple_env
+    ep = env.endpoints.create_start(
+        branch_name="main",
+        endpoint_id="primary",
+        config_lines=[
+            "max_wal_size=32MB",
+            "min_wal_size=32MB",
+            "neon.logical_replication_max_snap_files=10000",
+        ],
+    )
+
+    con = ep.connect()
+    cur = con.cursor()
+    cur.execute("CREATE TABLE t(pk bigint primary key, payload text)")
+    cur.execute("ALTER TABLE t ALTER payload SET STORAGE external")
+    cur.execute("select pg_create_logical_replication_slot('myslot', 'test_decoding', false, true)")
+    cur.execute("insert into t values (generate_series(1,100000),repeat('?',10000))")
+
+    ep.stop("fast")
+    ep.start()
+    con = ep.connect()
+    cur = con.cursor()
+    cur.execute("select pg_replication_slot_advance('myslot', pg_current_wal_insert_lsn())")

vendor/revisions.json

@@ -1,18 +1,18 @@
 {
   "v17": [
     "17.2",
-    "a05dc1378dd822276dc99cb5e888f905d3527597"
+    "e5d795a1a0c25da907176d37c905badab70e00c0"
   ],
   "v16": [
     "16.6",
-    "b7e9ac3eb9c5f43c443ebc76ddf06d5038c9bb34"
+    "3cc152ae2d17b19679c7102486bdb94677705c02"
   ],
   "v15": [
     "15.10",
-    "29bf1f04a5628618b4c7972fed6f87065e3750ce"
+    "544620db4ca6945be4f1f686a7fbd2cdfb0bf96f"
   ],
   "v14": [
     "14.15",
-    "e54af3504513b1f44c0e0f68791a0d6d4210e948"
+    "aeecd27b1f0775b606409d1cbb9c8aa9853a82af"
   ]
 }