From 7e6252c3d53255044ba7d9d5dbda63ed4917a657 Mon Sep 17 00:00:00 2001 From: Konstantin Knizhnik Date: Tue, 15 Aug 2023 15:59:43 +0300 Subject: [PATCH] Rewrite handling of XlHeapDelete XlHeapLock --- libs/postgres_ffi/src/pg_constants.rs | 3 + pageserver/src/walrecord.rs | 82 ++++++++++++++++----------- 2 files changed, 51 insertions(+), 34 deletions(-) diff --git a/libs/postgres_ffi/src/pg_constants.rs b/libs/postgres_ffi/src/pg_constants.rs index 2376e1fd9f..a1fa8598c7 100644 --- a/libs/postgres_ffi/src/pg_constants.rs +++ b/libs/postgres_ffi/src/pg_constants.rs @@ -150,6 +150,9 @@ 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; diff --git a/pageserver/src/walrecord.rs b/pageserver/src/walrecord.rs index df337d7a0a..e5211d732a 100644 --- a/pageserver/src/walrecord.rs +++ b/pageserver/src/walrecord.rs @@ -270,24 +270,31 @@ pub struct XlHeapDelete { impl XlHeapDelete { pub fn decode(buf: &mut Bytes) -> XlHeapDelete { - 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(), - } + let xmax = buf.get_u32_le(); + let offnum = buf.get_u16_le(); + let _padding; + let t_cid; + let neon_format; + if buf.remaining() == pg_constants::SIZE_OF_HEAP_DELETE { + // Neon format + _padding = buf.get_u16_le(); + t_cid = buf.get_u32_le(); + neon_format = true; } 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(), - } + _padding = 0; + t_cid = 0; + neon_format = false; + } + let infobits_set = buf.get_u8(); + let flags = buf.get_u8(); + assert!(((flags & pg_constants::XLH_DELETE_STORE_CID) == 0) ^ neon_format); + XlHeapDelete { + xmax, + offnum, + _padding, + t_cid, + infobits_set, + flags, } } } @@ -305,24 +312,31 @@ pub struct XlHeapLock { 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(), - } + let locking_xid = buf.get_u32_le(); + let offnum = buf.get_u16_le(); + let _padding; + let t_cid; + let neon_format; + if buf.remaining() == pg_constants::SIZE_OF_HEAP_LOCK { + // Neon format + _padding = buf.get_u16_le(); + t_cid = buf.get_u32_le(); + neon_format = true; } 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(), - } + _padding = 0; + t_cid = 0; + neon_format = false; + } + 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, } } }