Fix decoiding of HeapDelete/HeapLock commands

libs/postgres_ffi/src/pg_constants.rs

@@ -145,6 +145,13 @@ 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;
+
+pub const SIZE_OF_HEAP_LOCK: usize = 14;
+pub const SIZE_OF_HEAP_DELETE: usize = 14;
 
 // From replication/message.h
 pub const XLOG_LOGICAL_MESSAGE: u8 = 0x00;

pageserver/src/walingest.rs

@@ -450,15 +450,6 @@ impl<'a> WalIngest<'a> {
             let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
             if info == pg_constants::XLOG_HEAP2_MULTI_INSERT {
                 let xlrec = XlHeapMultiInsert::decode(buf);
-
-                let offset_array_len = if decoded.xl_info & pg_constants::XLOG_HEAP_INIT_PAGE > 0 {
-                    // the offsets array is omitted if XLOG_HEAP_INIT_PAGE is set
-                    0
-                } else {
-                    std::mem::size_of::<u16>() * xlrec.ntuples as usize
-                };
-                assert_eq!(offset_array_len, buf.remaining());
-
                 if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
                     new_heap_blkno = Some(decoded.blocks[0].blkno);
                 }

pageserver/src/walrecord.rs

@@ -270,13 +270,67 @@ pub struct XlHeapDelete {
 
 impl XlHeapDelete {
     pub fn decode(buf: &mut Bytes) -> XlHeapDelete {
+        let neon_format = buf.remaining() == pg_constants::SIZE_OF_HEAP_DELETE;
+        let xmax = buf.get_u32_le();
+        let offnum = buf.get_u16_le();
+        let _padding;
+        let t_cid;
+        if neon_format {
+            _padding = buf.get_u16_le();
+            t_cid = buf.get_u32_le();
+        } else {
+            _padding = 0;
+            t_cid = 0;
+        }
+        let infobits_set = buf.get_u8();
+        let flags = buf.get_u8();
+        assert!(((flags & pg_constants::XLH_DELETE_STORE_CID) == 0) ^ neon_format);
         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(),
+            xmax,
+            offnum,
+            _padding,
+            t_cid,
+            infobits_set,
+            flags,
+        }
+    }
+}
+
+#[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 {
+        let neon_format = buf.remaining() == pg_constants::SIZE_OF_HEAP_LOCK;
+        let locking_xid = buf.get_u32_le();
+        let offnum = buf.get_u16_le();
+        let _padding;
+        let t_cid;
+        if neon_format {
+            _padding = buf.get_u16_le();
+            t_cid = buf.get_u32_le();
+        } else {
+            _padding = 0;
+            t_cid = 0;
+        }
+        let infobits_set = buf.get_u8();
+        let flags = buf.get_u8();
+        assert!(((flags & pg_constants::XLH_LOCK_STORE_CID) == 0) ^ neon_format);
+        XlHeapLock {
+            locking_xid,
+            offnum,
+            _padding,
+            t_cid,
+            infobits_set,
+            flags,
         }
     }
 }
@@ -295,12 +349,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(),
         }

vendor/revisions.json

@@ -1,4 +1,4 @@
 {
-    "postgres-v15": "026d6b093d49e25cec44dd04598152329ceac027",
-    "postgres-v14": "5d5cfee12783f0989a9c9fe13bb40b5585812568"
+    "postgres-v15": "2c76abf4d54b4d9e7ef5f4a86184f15747fb7138",
+    "postgres-v14": "71126b905c5000e1a12d96640c94df8c3ec7384a"
 }