From d8735aa12ab1e00338c2a32d9f0e7cf5d479d0be Mon Sep 17 00:00:00 2001
From: Konstantin Knizhnik <knizhnik@neon.tech>
Date: Tue, 15 Aug 2023 14:32:03 +0300
Subject: [PATCH] Handle both Vanilla and Neon WAL formats

---
 libs/postgres_ffi/src/pg_constants.rs |  4 ++
 pageserver/src/walrecord.rs           | 79 +++++++++++++++++++++++----
 vendor/postgres-v14                   |  2 +-
 vendor/postgres-v15                   |  2 +-
 4 files changed, 73 insertions(+), 14 deletions(-)

diff --git a/libs/postgres_ffi/src/pg_constants.rs b/libs/postgres_ffi/src/pg_constants.rs
index 9c39b46cc1..2376e1fd9f 100644
--- a/libs/postgres_ffi/src/pg_constants.rs
+++ b/libs/postgres_ffi/src/pg_constants.rs
@@ -145,6 +145,10 @@ pub const XLH_INSERT_ALL_VISIBLE_CLEARED: u8 = (1 << 0) as u8;
 pub const XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED: u8 = (1 << 0) as u8;
 pub const XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED: u8 = (1 << 1) as u8;
 pub const XLH_DELETE_ALL_VISIBLE_CLEARED: u8 = (1 << 0) as u8;
+pub const XLH_INSERT_STORE_CID: u8 = (1 << 7) as u8;
+pub const XLH_UPDATE_STORE_CID: u8 = (1 << 7) as u8;
+pub const XLH_DELETE_STORE_CID: u8 = (1 << 7) as u8;
+pub const XLH_LOCK_STORE_CID: u8 = (1 << 7) as u8;
 
 // From replication/message.h
 pub const XLOG_LOGICAL_MESSAGE: u8 = 0x00;
diff --git a/pageserver/src/walrecord.rs b/pageserver/src/walrecord.rs
index 1a34168fed..df337d7a0a 100644
--- a/pageserver/src/walrecord.rs
+++ b/pageserver/src/walrecord.rs
@@ -270,13 +270,59 @@ pub struct XlHeapDelete {
 
 impl XlHeapDelete {
     pub fn decode(buf: &mut Bytes) -> XlHeapDelete {
-        XlHeapDelete {
-            xmax: buf.get_u32_le(),
-            offnum: buf.get_u16_le(),
-            _padding: buf.get_u16_le(),
-            t_cid: buf.get_u32_le(),
-            infobits_set: buf.get_u8(),
-            flags: buf.get_u8(),
+        if buf.remaining() == 14 {
+            XlHeapDelete {
+                xmax: buf.get_u32_le(),
+                offnum: buf.get_u16_le(),
+                _padding: buf.get_u16_le(),
+                t_cid: buf.get_u32_le(),
+                infobits_set: buf.get_u8(),
+                flags: buf.get_u8(),
+            }
+        } else {
+            XlHeapDelete {
+                xmax: buf.get_u32_le(),
+                offnum: buf.get_u16_le(),
+                _padding: 0,
+                t_cid: 0,
+                infobits_set: buf.get_u8(),
+                flags: buf.get_u8(),
+            }
+        }
+    }
+}
+
+#[repr(C)]
+#[derive(Debug)]
+pub struct XlHeapLock {
+    pub locking_xid: TransactionId,
+    pub offnum: OffsetNumber,
+    pub _padding: u16,
+    pub t_cid: u32,
+    pub infobits_set: u8,
+    pub flags: u8,
+}
+
+impl XlHeapLock {
+    pub fn decode(buf: &mut Bytes) -> XlHeapLock {
+        if buf.remaining() == 14 {
+            XlHeapLock {
+                locking_xid: buf.get_u32_le(),
+                offnum: buf.get_u16_le(),
+                _padding: buf.get_u16_le(),
+                t_cid: buf.get_u32_le(),
+                infobits_set: buf.get_u8(),
+                flags: buf.get_u8(),
+            }
+        } else {
+            XlHeapLock {
+                locking_xid: buf.get_u32_le(),
+                offnum: buf.get_u16_le(),
+                _padding: 0,
+                t_cid: 0,
+                infobits_set: buf.get_u8(),
+                flags: buf.get_u8(),
+            }
         }
     }
 }
@@ -295,12 +341,21 @@ pub struct XlHeapUpdate {
 
 impl XlHeapUpdate {
     pub fn decode(buf: &mut Bytes) -> XlHeapUpdate {
+        let old_xmax = buf.get_u32_le();
+        let old_offnum = buf.get_u16_le();
+        let old_infobits_set = buf.get_u8();
+        let flags = buf.get_u8();
+        let t_cid = if (flags & pg_constants::XLH_UPDATE_STORE_CID) != 0 {
+            buf.get_u32()
+        } else {
+            0
+        };
         XlHeapUpdate {
-            old_xmax: buf.get_u32_le(),
-            old_offnum: buf.get_u16_le(),
-            old_infobits_set: buf.get_u8(),
-            flags: buf.get_u8(),
-            t_cid: buf.get_u32(),
+            old_xmax,
+            old_offnum,
+            old_infobits_set,
+            flags,
+            t_cid,
             new_xmax: buf.get_u32_le(),
             new_offnum: buf.get_u16_le(),
         }
diff --git a/vendor/postgres-v14 b/vendor/postgres-v14
index 5d5cfee127..d5df566c20 160000
--- a/vendor/postgres-v14
+++ b/vendor/postgres-v14
@@ -1 +1 @@
-Subproject commit 5d5cfee12783f0989a9c9fe13bb40b5585812568
+Subproject commit d5df566c20290f09cfb8884f52a4085e32c68d2e
diff --git a/vendor/postgres-v15 b/vendor/postgres-v15
index 026d6b093d..f6ea27beb8 160000
--- a/vendor/postgres-v15
+++ b/vendor/postgres-v15
@@ -1 +1 @@
-Subproject commit 026d6b093d49e25cec44dd04598152329ceac027
+Subproject commit f6ea27beb8ad640ce7c78053803914eab914023a