LCOV - code coverage report
Current view: top level - libs/postgres_ffi/src - walrecord.rs (source / functions) Coverage Total Hit
Test: 8b13a09a5c233d98abd4a0d3e59157e7db16d6fd.info Lines: 37.8 % 606 229
Test Date: 2024-11-21 10:53:51 Functions: 7.0 % 115 8

            Line data    Source code
       1              : //! This module houses types used in decoding of PG WAL
       2              : //! records.
       3              : //!
       4              : //! TODO: Generate separate types for each supported PG version
       5              : 
       6              : use crate::pg_constants;
       7              : use crate::XLogRecord;
       8              : use crate::{
       9              :     BlockNumber, MultiXactId, MultiXactOffset, MultiXactStatus, Oid, RepOriginId, TimestampTz,
      10              :     TransactionId,
      11              : };
      12              : use crate::{BLCKSZ, XLOG_SIZE_OF_XLOG_RECORD};
      13              : use bytes::{Buf, Bytes};
      14              : use serde::{Deserialize, Serialize};
      15              : use utils::bin_ser::DeserializeError;
      16              : use utils::lsn::Lsn;
      17              : 
      18              : #[repr(C)]
      19            0 : #[derive(Debug, Serialize, Deserialize)]
      20              : pub struct XlMultiXactCreate {
      21              :     pub mid: MultiXactId,
      22              :     /* new MultiXact's ID */
      23              :     pub moff: MultiXactOffset,
      24              :     /* its starting offset in members file */
      25              :     pub nmembers: u32,
      26              :     /* number of member XIDs */
      27              :     pub members: Vec<MultiXactMember>,
      28              : }
      29              : 
      30              : impl XlMultiXactCreate {
      31            0 :     pub fn decode(buf: &mut Bytes) -> XlMultiXactCreate {
      32            0 :         let mid = buf.get_u32_le();
      33            0 :         let moff = buf.get_u32_le();
      34            0 :         let nmembers = buf.get_u32_le();
      35            0 :         let mut members = Vec::new();
      36            0 :         for _ in 0..nmembers {
      37            0 :             members.push(MultiXactMember::decode(buf));
      38            0 :         }
      39            0 :         XlMultiXactCreate {
      40            0 :             mid,
      41            0 :             moff,
      42            0 :             nmembers,
      43            0 :             members,
      44            0 :         }
      45            0 :     }
      46              : }
      47              : 
      48              : #[repr(C)]
      49            0 : #[derive(Debug, Serialize, Deserialize)]
      50              : pub struct XlMultiXactTruncate {
      51              :     pub oldest_multi_db: Oid,
      52              :     /* to-be-truncated range of multixact offsets */
      53              :     pub start_trunc_off: MultiXactId,
      54              :     /* just for completeness' sake */
      55              :     pub end_trunc_off: MultiXactId,
      56              : 
      57              :     /* to-be-truncated range of multixact members */
      58              :     pub start_trunc_memb: MultiXactOffset,
      59              :     pub end_trunc_memb: MultiXactOffset,
      60              : }
      61              : 
      62              : impl XlMultiXactTruncate {
      63            0 :     pub fn decode(buf: &mut Bytes) -> XlMultiXactTruncate {
      64            0 :         XlMultiXactTruncate {
      65            0 :             oldest_multi_db: buf.get_u32_le(),
      66            0 :             start_trunc_off: buf.get_u32_le(),
      67            0 :             end_trunc_off: buf.get_u32_le(),
      68            0 :             start_trunc_memb: buf.get_u32_le(),
      69            0 :             end_trunc_memb: buf.get_u32_le(),
      70            0 :         }
      71            0 :     }
      72              : }
      73              : 
      74              : #[repr(C)]
      75            0 : #[derive(Debug, Serialize, Deserialize)]
      76              : pub struct XlRelmapUpdate {
      77              :     pub dbid: Oid,   /* database ID, or 0 for shared map */
      78              :     pub tsid: Oid,   /* database's tablespace, or pg_global */
      79              :     pub nbytes: i32, /* size of relmap data */
      80              : }
      81              : 
      82              : impl XlRelmapUpdate {
      83            0 :     pub fn decode(buf: &mut Bytes) -> XlRelmapUpdate {
      84            0 :         XlRelmapUpdate {
      85            0 :             dbid: buf.get_u32_le(),
      86            0 :             tsid: buf.get_u32_le(),
      87            0 :             nbytes: buf.get_i32_le(),
      88            0 :         }
      89            0 :     }
      90              : }
      91              : 
      92              : #[repr(C)]
      93            0 : #[derive(Debug, Serialize, Deserialize)]
      94              : pub struct XlReploriginDrop {
      95              :     pub node_id: RepOriginId,
      96              : }
      97              : 
      98              : impl XlReploriginDrop {
      99            0 :     pub fn decode(buf: &mut Bytes) -> XlReploriginDrop {
     100            0 :         XlReploriginDrop {
     101            0 :             node_id: buf.get_u16_le(),
     102            0 :         }
     103            0 :     }
     104              : }
     105              : 
     106              : #[repr(C)]
     107            0 : #[derive(Debug, Serialize, Deserialize)]
     108              : pub struct XlReploriginSet {
     109              :     pub remote_lsn: Lsn,
     110              :     pub node_id: RepOriginId,
     111              : }
     112              : 
     113              : impl XlReploriginSet {
     114            0 :     pub fn decode(buf: &mut Bytes) -> XlReploriginSet {
     115            0 :         XlReploriginSet {
     116            0 :             remote_lsn: Lsn(buf.get_u64_le()),
     117            0 :             node_id: buf.get_u16_le(),
     118            0 :         }
     119            0 :     }
     120              : }
     121              : 
     122              : #[repr(C)]
     123            0 : #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
     124              : pub struct RelFileNode {
     125              :     pub spcnode: Oid, /* tablespace */
     126              :     pub dbnode: Oid,  /* database */
     127              :     pub relnode: Oid, /* relation */
     128              : }
     129              : 
     130              : #[repr(C)]
     131            0 : #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
     132              : pub struct MultiXactMember {
     133              :     pub xid: TransactionId,
     134              :     pub status: MultiXactStatus,
     135              : }
     136              : 
     137              : impl MultiXactMember {
     138            0 :     pub fn decode(buf: &mut Bytes) -> MultiXactMember {
     139            0 :         MultiXactMember {
     140            0 :             xid: buf.get_u32_le(),
     141            0 :             status: buf.get_u32_le(),
     142            0 :         }
     143            0 :     }
     144              : }
     145              : 
     146              : /// DecodedBkpBlock represents per-page data contained in a WAL record.
     147              : #[derive(Default)]
     148              : pub struct DecodedBkpBlock {
     149              :     /* Is this block ref in use? */
     150              :     //in_use: bool,
     151              : 
     152              :     /* Identify the block this refers to */
     153              :     pub rnode_spcnode: u32,
     154              :     pub rnode_dbnode: u32,
     155              :     pub rnode_relnode: u32,
     156              :     // Note that we have a few special forknum values for non-rel files.
     157              :     pub forknum: u8,
     158              :     pub blkno: u32,
     159              : 
     160              :     /* copy of the fork_flags field from the XLogRecordBlockHeader */
     161              :     pub flags: u8,
     162              : 
     163              :     /* Information on full-page image, if any */
     164              :     pub has_image: bool,
     165              :     /* has image, even for consistency checking */
     166              :     pub apply_image: bool,
     167              :     /* has image that should be restored */
     168              :     pub will_init: bool,
     169              :     /* record doesn't need previous page version to apply */
     170              :     //char         *bkp_image;
     171              :     pub hole_offset: u16,
     172              :     pub hole_length: u16,
     173              :     pub bimg_offset: u32,
     174              :     pub bimg_len: u16,
     175              :     pub bimg_info: u8,
     176              : 
     177              :     /* Buffer holding the rmgr-specific data associated with this block */
     178              :     has_data: bool,
     179              :     data_len: u16,
     180              : }
     181              : 
     182              : impl DecodedBkpBlock {
     183       145642 :     pub fn new() -> DecodedBkpBlock {
     184       145642 :         Default::default()
     185       145642 :     }
     186              : }
     187              : 
     188              : #[derive(Default)]
     189              : pub struct DecodedWALRecord {
     190              :     pub xl_xid: TransactionId,
     191              :     pub xl_info: u8,
     192              :     pub xl_rmid: u8,
     193              :     pub record: Bytes, // raw XLogRecord
     194              : 
     195              :     pub blocks: Vec<DecodedBkpBlock>,
     196              :     pub main_data_offset: usize,
     197              :     pub origin_id: u16,
     198              : }
     199              : 
     200              : impl DecodedWALRecord {
     201              :     /// Check if this WAL record represents a legacy "copy" database creation, which populates new relations
     202              :     /// by reading other existing relations' data blocks.  This is more complex to apply than new-style database
     203              :     /// creations which simply include all the desired blocks in the WAL, so we need a helper function to detect this case.
     204       145852 :     pub fn is_dbase_create_copy(&self, pg_version: u32) -> bool {
     205       145852 :         if self.xl_rmid == pg_constants::RM_DBASE_ID {
     206            0 :             let info = self.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
     207            0 :             match pg_version {
     208              :                 14 => {
     209              :                     // Postgres 14 database creations are always the legacy kind
     210            0 :                     info == crate::v14::bindings::XLOG_DBASE_CREATE
     211              :                 }
     212            0 :                 15 => info == crate::v15::bindings::XLOG_DBASE_CREATE_FILE_COPY,
     213            0 :                 16 => info == crate::v16::bindings::XLOG_DBASE_CREATE_FILE_COPY,
     214            0 :                 17 => info == crate::v17::bindings::XLOG_DBASE_CREATE_FILE_COPY,
     215              :                 _ => {
     216            0 :                     panic!("Unsupported postgres version {pg_version}")
     217              :                 }
     218              :             }
     219              :         } else {
     220       145852 :             false
     221              :         }
     222       145852 :     }
     223              : }
     224              : 
     225              : /// Main routine to decode a WAL record and figure out which blocks are modified
     226              : //
     227              : // See xlogrecord.h for details
     228              : // The overall layout of an XLOG record is:
     229              : //              Fixed-size header (XLogRecord struct)
     230              : //      XLogRecordBlockHeader struct
     231              : //          If pg_constants::BKPBLOCK_HAS_IMAGE, an XLogRecordBlockImageHeader struct follows
     232              : //                 If pg_constants::BKPIMAGE_HAS_HOLE and pg_constants::BKPIMAGE_IS_COMPRESSED, an
     233              : //                 XLogRecordBlockCompressHeader struct follows.
     234              : //          If pg_constants::BKPBLOCK_SAME_REL is not set, a RelFileNode follows
     235              : //          BlockNumber follows
     236              : //      XLogRecordBlockHeader struct
     237              : //      ...
     238              : //      XLogRecordDataHeader[Short|Long] struct
     239              : //      block data
     240              : //      block data
     241              : //      ...
     242              : //      main data
     243              : //
     244              : //
     245              : // For performance reasons, the caller provides the DecodedWALRecord struct and the function just fills it in.
     246              : // It would be more natural for this function to return a DecodedWALRecord as return value,
     247              : // but reusing the caller-supplied struct avoids an allocation.
     248              : // This code is in the hot path for digesting incoming WAL, and is very performance sensitive.
     249              : //
     250       145852 : pub fn decode_wal_record(
     251       145852 :     record: Bytes,
     252       145852 :     decoded: &mut DecodedWALRecord,
     253       145852 :     pg_version: u32,
     254       145852 : ) -> anyhow::Result<()> {
     255       145852 :     let mut rnode_spcnode: u32 = 0;
     256       145852 :     let mut rnode_dbnode: u32 = 0;
     257       145852 :     let mut rnode_relnode: u32 = 0;
     258       145852 :     let mut got_rnode = false;
     259       145852 :     let mut origin_id: u16 = 0;
     260       145852 : 
     261       145852 :     let mut buf = record.clone();
     262              : 
     263              :     // 1. Parse XLogRecord struct
     264              : 
     265              :     // FIXME: assume little-endian here
     266       145852 :     let xlogrec = XLogRecord::from_bytes(&mut buf)?;
     267              : 
     268       145852 :     tracing::trace!(
     269            0 :         "decode_wal_record xl_rmid = {} xl_info = {}",
     270              :         xlogrec.xl_rmid,
     271              :         xlogrec.xl_info
     272              :     );
     273              : 
     274       145852 :     let remaining: usize = xlogrec.xl_tot_len as usize - XLOG_SIZE_OF_XLOG_RECORD;
     275       145852 : 
     276       145852 :     if buf.remaining() != remaining {
     277            0 :         //TODO error
     278       145852 :     }
     279              : 
     280       145852 :     let mut max_block_id = 0;
     281       145852 :     let mut blocks_total_len: u32 = 0;
     282       145852 :     let mut main_data_len = 0;
     283       145852 :     let mut datatotal: u32 = 0;
     284       145852 :     decoded.blocks.clear();
     285              : 
     286              :     // 2. Decode the headers.
     287              :     // XLogRecordBlockHeaders if any,
     288              :     // XLogRecordDataHeader[Short|Long]
     289       437322 :     while buf.remaining() > datatotal as usize {
     290       291470 :         let block_id = buf.get_u8();
     291       291470 : 
     292       291470 :         match block_id {
     293       145812 :             pg_constants::XLR_BLOCK_ID_DATA_SHORT => {
     294       145812 :                 /* XLogRecordDataHeaderShort */
     295       145812 :                 main_data_len = buf.get_u8() as u32;
     296       145812 :                 datatotal += main_data_len;
     297       145812 :             }
     298              : 
     299           16 :             pg_constants::XLR_BLOCK_ID_DATA_LONG => {
     300           16 :                 /* XLogRecordDataHeaderLong */
     301           16 :                 main_data_len = buf.get_u32_le();
     302           16 :                 datatotal += main_data_len;
     303           16 :             }
     304              : 
     305            0 :             pg_constants::XLR_BLOCK_ID_ORIGIN => {
     306            0 :                 // RepOriginId is uint16
     307            0 :                 origin_id = buf.get_u16_le();
     308            0 :             }
     309              : 
     310            0 :             pg_constants::XLR_BLOCK_ID_TOPLEVEL_XID => {
     311            0 :                 // TransactionId is uint32
     312            0 :                 buf.advance(4);
     313            0 :             }
     314              : 
     315       145642 :             0..=pg_constants::XLR_MAX_BLOCK_ID => {
     316              :                 /* XLogRecordBlockHeader */
     317       145642 :                 let mut blk = DecodedBkpBlock::new();
     318       145642 : 
     319       145642 :                 if block_id <= max_block_id {
     320       145642 :                     // TODO
     321       145642 :                     //report_invalid_record(state,
     322       145642 :                     //                    "out-of-order block_id %u at %X/%X",
     323       145642 :                     //                    block_id,
     324       145642 :                     //                    (uint32) (state->ReadRecPtr >> 32),
     325       145642 :                     //                    (uint32) state->ReadRecPtr);
     326       145642 :                     //    goto err;
     327       145642 :                 }
     328       145642 :                 max_block_id = block_id;
     329       145642 : 
     330       145642 :                 let fork_flags: u8 = buf.get_u8();
     331       145642 :                 blk.forknum = fork_flags & pg_constants::BKPBLOCK_FORK_MASK;
     332       145642 :                 blk.flags = fork_flags;
     333       145642 :                 blk.has_image = (fork_flags & pg_constants::BKPBLOCK_HAS_IMAGE) != 0;
     334       145642 :                 blk.has_data = (fork_flags & pg_constants::BKPBLOCK_HAS_DATA) != 0;
     335       145642 :                 blk.will_init = (fork_flags & pg_constants::BKPBLOCK_WILL_INIT) != 0;
     336       145642 :                 blk.data_len = buf.get_u16_le();
     337       145642 : 
     338       145642 :                 /* TODO cross-check that the HAS_DATA flag is set iff data_length > 0 */
     339       145642 : 
     340       145642 :                 datatotal += blk.data_len as u32;
     341       145642 :                 blocks_total_len += blk.data_len as u32;
     342       145642 : 
     343       145642 :                 if blk.has_image {
     344           60 :                     blk.bimg_len = buf.get_u16_le();
     345           60 :                     blk.hole_offset = buf.get_u16_le();
     346           60 :                     blk.bimg_info = buf.get_u8();
     347              : 
     348            0 :                     blk.apply_image = dispatch_pgversion!(
     349           60 :                         pg_version,
     350            0 :                         (blk.bimg_info & pgv::bindings::BKPIMAGE_APPLY) != 0
     351              :                     );
     352              : 
     353           60 :                     let blk_img_is_compressed =
     354           60 :                         crate::bkpimage_is_compressed(blk.bimg_info, pg_version);
     355           60 : 
     356           60 :                     if blk_img_is_compressed {
     357            0 :                         tracing::debug!("compressed block image , pg_version = {}", pg_version);
     358           60 :                     }
     359              : 
     360           60 :                     if blk_img_is_compressed {
     361            0 :                         if blk.bimg_info & pg_constants::BKPIMAGE_HAS_HOLE != 0 {
     362            0 :                             blk.hole_length = buf.get_u16_le();
     363            0 :                         } else {
     364            0 :                             blk.hole_length = 0;
     365            0 :                         }
     366           60 :                     } else {
     367           60 :                         blk.hole_length = BLCKSZ - blk.bimg_len;
     368           60 :                     }
     369           60 :                     datatotal += blk.bimg_len as u32;
     370           60 :                     blocks_total_len += blk.bimg_len as u32;
     371           60 : 
     372           60 :                     /*
     373           60 :                      * cross-check that hole_offset > 0, hole_length > 0 and
     374           60 :                      * bimg_len < BLCKSZ if the HAS_HOLE flag is set.
     375           60 :                      */
     376           60 :                     if blk.bimg_info & pg_constants::BKPIMAGE_HAS_HOLE != 0
     377           36 :                         && (blk.hole_offset == 0 || blk.hole_length == 0 || blk.bimg_len == BLCKSZ)
     378            0 :                     {
     379            0 :                         // TODO
     380            0 :                         /*
     381            0 :                         report_invalid_record(state,
     382            0 :                                       "pg_constants::BKPIMAGE_HAS_HOLE set, but hole offset %u length %u block image length %u at %X/%X",
     383            0 :                                       (unsigned int) blk->hole_offset,
     384            0 :                                       (unsigned int) blk->hole_length,
     385            0 :                                       (unsigned int) blk->bimg_len,
     386            0 :                                       (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr);
     387            0 :                         goto err;
     388            0 :                                      */
     389           60 :                     }
     390              : 
     391              :                     /*
     392              :                      * cross-check that hole_offset == 0 and hole_length == 0 if
     393              :                      * the HAS_HOLE flag is not set.
     394              :                      */
     395           60 :                     if blk.bimg_info & pg_constants::BKPIMAGE_HAS_HOLE == 0
     396           24 :                         && (blk.hole_offset != 0 || blk.hole_length != 0)
     397            0 :                     {
     398            0 :                         // TODO
     399            0 :                         /*
     400            0 :                         report_invalid_record(state,
     401            0 :                                       "pg_constants::BKPIMAGE_HAS_HOLE not set, but hole offset %u length %u at %X/%X",
     402            0 :                                       (unsigned int) blk->hole_offset,
     403            0 :                                       (unsigned int) blk->hole_length,
     404            0 :                                       (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr);
     405            0 :                         goto err;
     406            0 :                                      */
     407           60 :                     }
     408              : 
     409              :                     /*
     410              :                      * cross-check that bimg_len < BLCKSZ if the IS_COMPRESSED
     411              :                      * flag is set.
     412              :                      */
     413           60 :                     if !blk_img_is_compressed && blk.bimg_len == BLCKSZ {
     414           24 :                         // TODO
     415           24 :                         /*
     416           24 :                         report_invalid_record(state,
     417           24 :                                       "pg_constants::BKPIMAGE_IS_COMPRESSED set, but block image length %u at %X/%X",
     418           24 :                                       (unsigned int) blk->bimg_len,
     419           24 :                                       (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr);
     420           24 :                         goto err;
     421           24 :                                      */
     422           36 :                     }
     423              : 
     424              :                     /*
     425              :                      * cross-check that bimg_len = BLCKSZ if neither HAS_HOLE nor
     426              :                      * IS_COMPRESSED flag is set.
     427              :                      */
     428           60 :                     if blk.bimg_info & pg_constants::BKPIMAGE_HAS_HOLE == 0
     429           24 :                         && !blk_img_is_compressed
     430           24 :                         && blk.bimg_len != BLCKSZ
     431            0 :                     {
     432            0 :                         // TODO
     433            0 :                         /*
     434            0 :                         report_invalid_record(state,
     435            0 :                                       "neither pg_constants::BKPIMAGE_HAS_HOLE nor pg_constants::BKPIMAGE_IS_COMPRESSED set, but block image length is %u at %X/%X",
     436            0 :                                       (unsigned int) blk->data_len,
     437            0 :                                       (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr);
     438            0 :                         goto err;
     439            0 :                                      */
     440           60 :                     }
     441       145582 :                 }
     442       145642 :                 if fork_flags & pg_constants::BKPBLOCK_SAME_REL == 0 {
     443       145642 :                     rnode_spcnode = buf.get_u32_le();
     444       145642 :                     rnode_dbnode = buf.get_u32_le();
     445       145642 :                     rnode_relnode = buf.get_u32_le();
     446       145642 :                     got_rnode = true;
     447       145642 :                 } else if !got_rnode {
     448            0 :                     // TODO
     449            0 :                     /*
     450            0 :                     report_invalid_record(state,
     451            0 :                                     "pg_constants::BKPBLOCK_SAME_REL set but no previous rel at %X/%X",
     452            0 :                                     (uint32) (state->ReadRecPtr >> 32), (uint32) state->ReadRecPtr);
     453            0 :                     goto err;           */
     454            0 :                 }
     455              : 
     456       145642 :                 blk.rnode_spcnode = rnode_spcnode;
     457       145642 :                 blk.rnode_dbnode = rnode_dbnode;
     458       145642 :                 blk.rnode_relnode = rnode_relnode;
     459       145642 : 
     460       145642 :                 blk.blkno = buf.get_u32_le();
     461       145642 :                 tracing::trace!(
     462            0 :                     "this record affects {}/{}/{} blk {}",
     463              :                     rnode_spcnode,
     464              :                     rnode_dbnode,
     465              :                     rnode_relnode,
     466              :                     blk.blkno
     467              :                 );
     468              : 
     469       145642 :                 decoded.blocks.push(blk);
     470              :             }
     471              : 
     472            0 :             _ => {
     473            0 :                 // TODO: invalid block_id
     474            0 :             }
     475              :         }
     476              :     }
     477              : 
     478              :     // 3. Decode blocks.
     479       145852 :     let mut ptr = record.len() - buf.remaining();
     480       145852 :     for blk in decoded.blocks.iter_mut() {
     481       145642 :         if blk.has_image {
     482           60 :             blk.bimg_offset = ptr as u32;
     483           60 :             ptr += blk.bimg_len as usize;
     484       145582 :         }
     485       145642 :         if blk.has_data {
     486       145582 :             ptr += blk.data_len as usize;
     487       145582 :         }
     488              :     }
     489              :     // We don't need them, so just skip blocks_total_len bytes
     490       145852 :     buf.advance(blocks_total_len as usize);
     491       145852 :     assert_eq!(ptr, record.len() - buf.remaining());
     492              : 
     493       145852 :     let main_data_offset = (xlogrec.xl_tot_len - main_data_len) as usize;
     494       145852 : 
     495       145852 :     // 4. Decode main_data
     496       145852 :     if main_data_len > 0 {
     497       145828 :         assert_eq!(buf.remaining(), main_data_len as usize);
     498           24 :     }
     499              : 
     500       145852 :     decoded.xl_xid = xlogrec.xl_xid;
     501       145852 :     decoded.xl_info = xlogrec.xl_info;
     502       145852 :     decoded.xl_rmid = xlogrec.xl_rmid;
     503       145852 :     decoded.record = record;
     504       145852 :     decoded.origin_id = origin_id;
     505       145852 :     decoded.main_data_offset = main_data_offset;
     506       145852 : 
     507       145852 :     Ok(())
     508       145852 : }
     509              : 
     510              : pub mod v14 {
     511              :     use crate::{OffsetNumber, TransactionId};
     512              :     use bytes::{Buf, Bytes};
     513              : 
     514              :     #[repr(C)]
     515              :     #[derive(Debug)]
     516              :     pub struct XlHeapInsert {
     517              :         pub offnum: OffsetNumber,
     518              :         pub flags: u8,
     519              :     }
     520              : 
     521              :     impl XlHeapInsert {
     522       145276 :         pub fn decode(buf: &mut Bytes) -> XlHeapInsert {
     523       145276 :             XlHeapInsert {
     524       145276 :                 offnum: buf.get_u16_le(),
     525       145276 :                 flags: buf.get_u8(),
     526       145276 :             }
     527       145276 :         }
     528              :     }
     529              : 
     530              :     #[repr(C)]
     531              :     #[derive(Debug)]
     532              :     pub struct XlHeapMultiInsert {
     533              :         pub flags: u8,
     534              :         pub _padding: u8,
     535              :         pub ntuples: u16,
     536              :     }
     537              : 
     538              :     impl XlHeapMultiInsert {
     539           42 :         pub fn decode(buf: &mut Bytes) -> XlHeapMultiInsert {
     540           42 :             XlHeapMultiInsert {
     541           42 :                 flags: buf.get_u8(),
     542           42 :                 _padding: buf.get_u8(),
     543           42 :                 ntuples: buf.get_u16_le(),
     544           42 :             }
     545           42 :         }
     546              :     }
     547              : 
     548              :     #[repr(C)]
     549              :     #[derive(Debug)]
     550              :     pub struct XlHeapDelete {
     551              :         pub xmax: TransactionId,
     552              :         pub offnum: OffsetNumber,
     553              :         pub _padding: u16,
     554              :         pub t_cid: u32,
     555              :         pub infobits_set: u8,
     556              :         pub flags: u8,
     557              :     }
     558              : 
     559              :     impl XlHeapDelete {
     560            0 :         pub fn decode(buf: &mut Bytes) -> XlHeapDelete {
     561            0 :             XlHeapDelete {
     562            0 :                 xmax: buf.get_u32_le(),
     563            0 :                 offnum: buf.get_u16_le(),
     564            0 :                 _padding: buf.get_u16_le(),
     565            0 :                 t_cid: buf.get_u32_le(),
     566            0 :                 infobits_set: buf.get_u8(),
     567            0 :                 flags: buf.get_u8(),
     568            0 :             }
     569            0 :         }
     570              :     }
     571              : 
     572              :     #[repr(C)]
     573              :     #[derive(Debug)]
     574              :     pub struct XlHeapUpdate {
     575              :         pub old_xmax: TransactionId,
     576              :         pub old_offnum: OffsetNumber,
     577              :         pub old_infobits_set: u8,
     578              :         pub flags: u8,
     579              :         pub t_cid: u32,
     580              :         pub new_xmax: TransactionId,
     581              :         pub new_offnum: OffsetNumber,
     582              :     }
     583              : 
     584              :     impl XlHeapUpdate {
     585            8 :         pub fn decode(buf: &mut Bytes) -> XlHeapUpdate {
     586            8 :             XlHeapUpdate {
     587            8 :                 old_xmax: buf.get_u32_le(),
     588            8 :                 old_offnum: buf.get_u16_le(),
     589            8 :                 old_infobits_set: buf.get_u8(),
     590            8 :                 flags: buf.get_u8(),
     591            8 :                 t_cid: buf.get_u32_le(),
     592            8 :                 new_xmax: buf.get_u32_le(),
     593            8 :                 new_offnum: buf.get_u16_le(),
     594            8 :             }
     595            8 :         }
     596              :     }
     597              : 
     598              :     #[repr(C)]
     599              :     #[derive(Debug)]
     600              :     pub struct XlHeapLock {
     601              :         pub locking_xid: TransactionId,
     602              :         pub offnum: OffsetNumber,
     603              :         pub _padding: u16,
     604              :         pub t_cid: u32,
     605              :         pub infobits_set: u8,
     606              :         pub flags: u8,
     607              :     }
     608              : 
     609              :     impl XlHeapLock {
     610            0 :         pub fn decode(buf: &mut Bytes) -> XlHeapLock {
     611            0 :             XlHeapLock {
     612            0 :                 locking_xid: buf.get_u32_le(),
     613            0 :                 offnum: buf.get_u16_le(),
     614            0 :                 _padding: buf.get_u16_le(),
     615            0 :                 t_cid: buf.get_u32_le(),
     616            0 :                 infobits_set: buf.get_u8(),
     617            0 :                 flags: buf.get_u8(),
     618            0 :             }
     619            0 :         }
     620              :     }
     621              : 
     622              :     #[repr(C)]
     623              :     #[derive(Debug)]
     624              :     pub struct XlHeapLockUpdated {
     625              :         pub xmax: TransactionId,
     626              :         pub offnum: OffsetNumber,
     627              :         pub infobits_set: u8,
     628              :         pub flags: u8,
     629              :     }
     630              : 
     631              :     impl XlHeapLockUpdated {
     632            0 :         pub fn decode(buf: &mut Bytes) -> XlHeapLockUpdated {
     633            0 :             XlHeapLockUpdated {
     634            0 :                 xmax: buf.get_u32_le(),
     635            0 :                 offnum: buf.get_u16_le(),
     636            0 :                 infobits_set: buf.get_u8(),
     637            0 :                 flags: buf.get_u8(),
     638            0 :             }
     639            0 :         }
     640              :     }
     641              : 
     642              :     #[repr(C)]
     643              :     #[derive(Debug)]
     644              :     pub struct XlParameterChange {
     645              :         pub max_connections: i32,
     646              :         pub max_worker_processes: i32,
     647              :         pub max_wal_senders: i32,
     648              :         pub max_prepared_xacts: i32,
     649              :         pub max_locks_per_xact: i32,
     650              :         pub wal_level: i32,
     651              :         pub wal_log_hints: bool,
     652              :         pub track_commit_timestamp: bool,
     653              :         pub _padding: [u8; 2],
     654              :     }
     655              : 
     656              :     impl XlParameterChange {
     657            0 :         pub fn decode(buf: &mut Bytes) -> XlParameterChange {
     658            0 :             XlParameterChange {
     659            0 :                 max_connections: buf.get_i32_le(),
     660            0 :                 max_worker_processes: buf.get_i32_le(),
     661            0 :                 max_wal_senders: buf.get_i32_le(),
     662            0 :                 max_prepared_xacts: buf.get_i32_le(),
     663            0 :                 max_locks_per_xact: buf.get_i32_le(),
     664            0 :                 wal_level: buf.get_i32_le(),
     665            0 :                 wal_log_hints: buf.get_u8() != 0,
     666            0 :                 track_commit_timestamp: buf.get_u8() != 0,
     667            0 :                 _padding: [buf.get_u8(), buf.get_u8()],
     668            0 :             }
     669            0 :         }
     670              :     }
     671              : }
     672              : 
     673              : pub mod v15 {
     674              :     pub use super::v14::{
     675              :         XlHeapDelete, XlHeapInsert, XlHeapLock, XlHeapLockUpdated, XlHeapMultiInsert, XlHeapUpdate,
     676              :         XlParameterChange,
     677              :     };
     678              : }
     679              : 
     680              : pub mod v16 {
     681              :     pub use super::v14::{XlHeapInsert, XlHeapLockUpdated, XlHeapMultiInsert, XlParameterChange};
     682              :     use crate::{OffsetNumber, TransactionId};
     683              :     use bytes::{Buf, Bytes};
     684              : 
     685              :     pub struct XlHeapDelete {
     686              :         pub xmax: TransactionId,
     687              :         pub offnum: OffsetNumber,
     688              :         pub infobits_set: u8,
     689              :         pub flags: u8,
     690              :     }
     691              : 
     692              :     impl XlHeapDelete {
     693            0 :         pub fn decode(buf: &mut Bytes) -> XlHeapDelete {
     694            0 :             XlHeapDelete {
     695            0 :                 xmax: buf.get_u32_le(),
     696            0 :                 offnum: buf.get_u16_le(),
     697            0 :                 infobits_set: buf.get_u8(),
     698            0 :                 flags: buf.get_u8(),
     699            0 :             }
     700            0 :         }
     701              :     }
     702              : 
     703              :     #[repr(C)]
     704              :     #[derive(Debug)]
     705              :     pub struct XlHeapUpdate {
     706              :         pub old_xmax: TransactionId,
     707              :         pub old_offnum: OffsetNumber,
     708              :         pub old_infobits_set: u8,
     709              :         pub flags: u8,
     710              :         pub new_xmax: TransactionId,
     711              :         pub new_offnum: OffsetNumber,
     712              :     }
     713              : 
     714              :     impl XlHeapUpdate {
     715            0 :         pub fn decode(buf: &mut Bytes) -> XlHeapUpdate {
     716            0 :             XlHeapUpdate {
     717            0 :                 old_xmax: buf.get_u32_le(),
     718            0 :                 old_offnum: buf.get_u16_le(),
     719            0 :                 old_infobits_set: buf.get_u8(),
     720            0 :                 flags: buf.get_u8(),
     721            0 :                 new_xmax: buf.get_u32_le(),
     722            0 :                 new_offnum: buf.get_u16_le(),
     723            0 :             }
     724            0 :         }
     725              :     }
     726              : 
     727              :     #[repr(C)]
     728              :     #[derive(Debug)]
     729              :     pub struct XlHeapLock {
     730              :         pub locking_xid: TransactionId,
     731              :         pub offnum: OffsetNumber,
     732              :         pub infobits_set: u8,
     733              :         pub flags: u8,
     734              :     }
     735              : 
     736              :     impl XlHeapLock {
     737            0 :         pub fn decode(buf: &mut Bytes) -> XlHeapLock {
     738            0 :             XlHeapLock {
     739            0 :                 locking_xid: buf.get_u32_le(),
     740            0 :                 offnum: buf.get_u16_le(),
     741            0 :                 infobits_set: buf.get_u8(),
     742            0 :                 flags: buf.get_u8(),
     743            0 :             }
     744            0 :         }
     745              :     }
     746              : 
     747              :     /* Since PG16, we have the Neon RMGR (RM_NEON_ID) to manage Neon-flavored WAL. */
     748              :     pub mod rm_neon {
     749              :         use crate::{OffsetNumber, TransactionId};
     750              :         use bytes::{Buf, Bytes};
     751              : 
     752              :         #[repr(C)]
     753              :         #[derive(Debug)]
     754              :         pub struct XlNeonHeapInsert {
     755              :             pub offnum: OffsetNumber,
     756              :             pub flags: u8,
     757              :         }
     758              : 
     759              :         impl XlNeonHeapInsert {
     760            0 :             pub fn decode(buf: &mut Bytes) -> XlNeonHeapInsert {
     761            0 :                 XlNeonHeapInsert {
     762            0 :                     offnum: buf.get_u16_le(),
     763            0 :                     flags: buf.get_u8(),
     764            0 :                 }
     765            0 :             }
     766              :         }
     767              : 
     768              :         #[repr(C)]
     769              :         #[derive(Debug)]
     770              :         pub struct XlNeonHeapMultiInsert {
     771              :             pub flags: u8,
     772              :             pub _padding: u8,
     773              :             pub ntuples: u16,
     774              :             pub t_cid: u32,
     775              :         }
     776              : 
     777              :         impl XlNeonHeapMultiInsert {
     778            0 :             pub fn decode(buf: &mut Bytes) -> XlNeonHeapMultiInsert {
     779            0 :                 XlNeonHeapMultiInsert {
     780            0 :                     flags: buf.get_u8(),
     781            0 :                     _padding: buf.get_u8(),
     782            0 :                     ntuples: buf.get_u16_le(),
     783            0 :                     t_cid: buf.get_u32_le(),
     784            0 :                 }
     785            0 :             }
     786              :         }
     787              : 
     788              :         #[repr(C)]
     789              :         #[derive(Debug)]
     790              :         pub struct XlNeonHeapDelete {
     791              :             pub xmax: TransactionId,
     792              :             pub offnum: OffsetNumber,
     793              :             pub infobits_set: u8,
     794              :             pub flags: u8,
     795              :             pub t_cid: u32,
     796              :         }
     797              : 
     798              :         impl XlNeonHeapDelete {
     799            0 :             pub fn decode(buf: &mut Bytes) -> XlNeonHeapDelete {
     800            0 :                 XlNeonHeapDelete {
     801            0 :                     xmax: buf.get_u32_le(),
     802            0 :                     offnum: buf.get_u16_le(),
     803            0 :                     infobits_set: buf.get_u8(),
     804            0 :                     flags: buf.get_u8(),
     805            0 :                     t_cid: buf.get_u32_le(),
     806            0 :                 }
     807            0 :             }
     808              :         }
     809              : 
     810              :         #[repr(C)]
     811              :         #[derive(Debug)]
     812              :         pub struct XlNeonHeapUpdate {
     813              :             pub old_xmax: TransactionId,
     814              :             pub old_offnum: OffsetNumber,
     815              :             pub old_infobits_set: u8,
     816              :             pub flags: u8,
     817              :             pub t_cid: u32,
     818              :             pub new_xmax: TransactionId,
     819              :             pub new_offnum: OffsetNumber,
     820              :         }
     821              : 
     822              :         impl XlNeonHeapUpdate {
     823            0 :             pub fn decode(buf: &mut Bytes) -> XlNeonHeapUpdate {
     824            0 :                 XlNeonHeapUpdate {
     825            0 :                     old_xmax: buf.get_u32_le(),
     826            0 :                     old_offnum: buf.get_u16_le(),
     827            0 :                     old_infobits_set: buf.get_u8(),
     828            0 :                     flags: buf.get_u8(),
     829            0 :                     t_cid: buf.get_u32(),
     830            0 :                     new_xmax: buf.get_u32_le(),
     831            0 :                     new_offnum: buf.get_u16_le(),
     832            0 :                 }
     833            0 :             }
     834              :         }
     835              : 
     836              :         #[repr(C)]
     837              :         #[derive(Debug)]
     838              :         pub struct XlNeonHeapLock {
     839              :             pub locking_xid: TransactionId,
     840              :             pub t_cid: u32,
     841              :             pub offnum: OffsetNumber,
     842              :             pub infobits_set: u8,
     843              :             pub flags: u8,
     844              :         }
     845              : 
     846              :         impl XlNeonHeapLock {
     847            0 :             pub fn decode(buf: &mut Bytes) -> XlNeonHeapLock {
     848            0 :                 XlNeonHeapLock {
     849            0 :                     locking_xid: buf.get_u32_le(),
     850            0 :                     t_cid: buf.get_u32_le(),
     851            0 :                     offnum: buf.get_u16_le(),
     852            0 :                     infobits_set: buf.get_u8(),
     853            0 :                     flags: buf.get_u8(),
     854            0 :                 }
     855            0 :             }
     856              :         }
     857              :     }
     858              : }
     859              : 
     860              : pub mod v17 {
     861              :     pub use super::v14::XlHeapLockUpdated;
     862              :     pub use crate::{TimeLineID, TimestampTz};
     863              :     use bytes::{Buf, Bytes};
     864              : 
     865              :     pub use super::v16::rm_neon;
     866              :     pub use super::v16::{
     867              :         XlHeapDelete, XlHeapInsert, XlHeapLock, XlHeapMultiInsert, XlHeapUpdate, XlParameterChange,
     868              :     };
     869              : 
     870              :     #[repr(C)]
     871              :     #[derive(Debug)]
     872              :     pub struct XlEndOfRecovery {
     873              :         pub end_time: TimestampTz,
     874              :         pub this_time_line_id: TimeLineID,
     875              :         pub prev_time_line_id: TimeLineID,
     876              :         pub wal_level: i32,
     877              :     }
     878              : 
     879              :     impl XlEndOfRecovery {
     880            0 :         pub fn decode(buf: &mut Bytes) -> XlEndOfRecovery {
     881            0 :             XlEndOfRecovery {
     882            0 :                 end_time: buf.get_i64_le(),
     883            0 :                 this_time_line_id: buf.get_u32_le(),
     884            0 :                 prev_time_line_id: buf.get_u32_le(),
     885            0 :                 wal_level: buf.get_i32_le(),
     886            0 :             }
     887            0 :         }
     888              :     }
     889              : }
     890              : 
     891              : #[repr(C)]
     892              : #[derive(Debug)]
     893              : pub struct XlSmgrCreate {
     894              :     pub rnode: RelFileNode,
     895              :     // FIXME: This is ForkNumber in storage_xlog.h. That's an enum. Does it have
     896              :     // well-defined size?
     897              :     pub forknum: u8,
     898              : }
     899              : 
     900              : impl XlSmgrCreate {
     901           16 :     pub fn decode(buf: &mut Bytes) -> XlSmgrCreate {
     902           16 :         XlSmgrCreate {
     903           16 :             rnode: RelFileNode {
     904           16 :                 spcnode: buf.get_u32_le(), /* tablespace */
     905           16 :                 dbnode: buf.get_u32_le(),  /* database */
     906           16 :                 relnode: buf.get_u32_le(), /* relation */
     907           16 :             },
     908           16 :             forknum: buf.get_u32_le() as u8,
     909           16 :         }
     910           16 :     }
     911              : }
     912              : 
     913              : #[repr(C)]
     914            0 : #[derive(Debug, Serialize, Deserialize)]
     915              : pub struct XlSmgrTruncate {
     916              :     pub blkno: BlockNumber,
     917              :     pub rnode: RelFileNode,
     918              :     pub flags: u32,
     919              : }
     920              : 
     921              : impl XlSmgrTruncate {
     922            0 :     pub fn decode(buf: &mut Bytes) -> XlSmgrTruncate {
     923            0 :         XlSmgrTruncate {
     924            0 :             blkno: buf.get_u32_le(),
     925            0 :             rnode: RelFileNode {
     926            0 :                 spcnode: buf.get_u32_le(), /* tablespace */
     927            0 :                 dbnode: buf.get_u32_le(),  /* database */
     928            0 :                 relnode: buf.get_u32_le(), /* relation */
     929            0 :             },
     930            0 :             flags: buf.get_u32_le(),
     931            0 :         }
     932            0 :     }
     933              : }
     934              : 
     935              : #[repr(C)]
     936              : #[derive(Debug)]
     937              : pub struct XlCreateDatabase {
     938              :     pub db_id: Oid,
     939              :     pub tablespace_id: Oid,
     940              :     pub src_db_id: Oid,
     941              :     pub src_tablespace_id: Oid,
     942              : }
     943              : 
     944              : impl XlCreateDatabase {
     945            0 :     pub fn decode(buf: &mut Bytes) -> XlCreateDatabase {
     946            0 :         XlCreateDatabase {
     947            0 :             db_id: buf.get_u32_le(),
     948            0 :             tablespace_id: buf.get_u32_le(),
     949            0 :             src_db_id: buf.get_u32_le(),
     950            0 :             src_tablespace_id: buf.get_u32_le(),
     951            0 :         }
     952            0 :     }
     953              : }
     954              : 
     955              : #[repr(C)]
     956              : #[derive(Debug)]
     957              : pub struct XlDropDatabase {
     958              :     pub db_id: Oid,
     959              :     pub n_tablespaces: Oid, /* number of tablespace IDs */
     960              :     pub tablespace_ids: Vec<Oid>,
     961              : }
     962              : 
     963              : impl XlDropDatabase {
     964            0 :     pub fn decode(buf: &mut Bytes) -> XlDropDatabase {
     965            0 :         let mut rec = XlDropDatabase {
     966            0 :             db_id: buf.get_u32_le(),
     967            0 :             n_tablespaces: buf.get_u32_le(),
     968            0 :             tablespace_ids: Vec::<Oid>::new(),
     969            0 :         };
     970              : 
     971            0 :         for _i in 0..rec.n_tablespaces {
     972            0 :             let id = buf.get_u32_le();
     973            0 :             rec.tablespace_ids.push(id);
     974            0 :         }
     975              : 
     976            0 :         rec
     977            0 :     }
     978              : }
     979              : 
     980              : ///
     981              : /// Note: Parsing some fields is missing, because they're not needed.
     982              : ///
     983              : /// This is similar to the xl_xact_parsed_commit and
     984              : /// xl_xact_parsed_abort structs in PostgreSQL, but we use the same
     985              : /// struct for commits and aborts.
     986              : ///
     987            0 : #[derive(Debug, Serialize, Deserialize)]
     988              : pub struct XlXactParsedRecord {
     989              :     pub xid: TransactionId,
     990              :     pub info: u8,
     991              :     pub xact_time: TimestampTz,
     992              :     pub xinfo: u32,
     993              : 
     994              :     pub db_id: Oid,
     995              :     /* MyDatabaseId */
     996              :     pub ts_id: Oid,
     997              :     /* MyDatabaseTableSpace */
     998              :     pub subxacts: Vec<TransactionId>,
     999              : 
    1000              :     pub xnodes: Vec<RelFileNode>,
    1001              :     pub origin_lsn: Lsn,
    1002              : }
    1003              : 
    1004              : impl XlXactParsedRecord {
    1005              :     /// Decode a XLOG_XACT_COMMIT/ABORT/COMMIT_PREPARED/ABORT_PREPARED
    1006              :     /// record. This should agree with the ParseCommitRecord and ParseAbortRecord
    1007              :     /// functions in PostgreSQL (in src/backend/access/rmgr/xactdesc.c)
    1008            8 :     pub fn decode(buf: &mut Bytes, mut xid: TransactionId, xl_info: u8) -> XlXactParsedRecord {
    1009            8 :         let info = xl_info & pg_constants::XLOG_XACT_OPMASK;
    1010            8 :         // The record starts with time of commit/abort
    1011            8 :         let xact_time = buf.get_i64_le();
    1012            8 :         let xinfo = if xl_info & pg_constants::XLOG_XACT_HAS_INFO != 0 {
    1013            8 :             buf.get_u32_le()
    1014              :         } else {
    1015            0 :             0
    1016              :         };
    1017              :         let db_id;
    1018              :         let ts_id;
    1019            8 :         if xinfo & pg_constants::XACT_XINFO_HAS_DBINFO != 0 {
    1020            8 :             db_id = buf.get_u32_le();
    1021            8 :             ts_id = buf.get_u32_le();
    1022            8 :         } else {
    1023            0 :             db_id = 0;
    1024            0 :             ts_id = 0;
    1025            0 :         }
    1026            8 :         let mut subxacts = Vec::<TransactionId>::new();
    1027            8 :         if xinfo & pg_constants::XACT_XINFO_HAS_SUBXACTS != 0 {
    1028            0 :             let nsubxacts = buf.get_i32_le();
    1029            0 :             for _i in 0..nsubxacts {
    1030            0 :                 let subxact = buf.get_u32_le();
    1031            0 :                 subxacts.push(subxact);
    1032            0 :             }
    1033            8 :         }
    1034            8 :         let mut xnodes = Vec::<RelFileNode>::new();
    1035            8 :         if xinfo & pg_constants::XACT_XINFO_HAS_RELFILENODES != 0 {
    1036            0 :             let nrels = buf.get_i32_le();
    1037            0 :             for _i in 0..nrels {
    1038            0 :                 let spcnode = buf.get_u32_le();
    1039            0 :                 let dbnode = buf.get_u32_le();
    1040            0 :                 let relnode = buf.get_u32_le();
    1041            0 :                 tracing::trace!(
    1042            0 :                     "XLOG_XACT_COMMIT relfilenode {}/{}/{}",
    1043              :                     spcnode,
    1044              :                     dbnode,
    1045              :                     relnode
    1046              :                 );
    1047            0 :                 xnodes.push(RelFileNode {
    1048            0 :                     spcnode,
    1049            0 :                     dbnode,
    1050            0 :                     relnode,
    1051            0 :                 });
    1052              :             }
    1053            8 :         }
    1054              : 
    1055            8 :         if xinfo & crate::v15::bindings::XACT_XINFO_HAS_DROPPED_STATS != 0 {
    1056            0 :             let nitems = buf.get_i32_le();
    1057            0 :             tracing::debug!(
    1058            0 :                 "XLOG_XACT_COMMIT-XACT_XINFO_HAS_DROPPED_STAT nitems {}",
    1059              :                 nitems
    1060              :             );
    1061            0 :             let sizeof_xl_xact_stats_item = 12;
    1062            0 :             buf.advance((nitems * sizeof_xl_xact_stats_item).try_into().unwrap());
    1063            8 :         }
    1064              : 
    1065            8 :         if xinfo & pg_constants::XACT_XINFO_HAS_INVALS != 0 {
    1066            8 :             let nmsgs = buf.get_i32_le();
    1067            8 :             let sizeof_shared_invalidation_message = 16;
    1068            8 :             buf.advance(
    1069            8 :                 (nmsgs * sizeof_shared_invalidation_message)
    1070            8 :                     .try_into()
    1071            8 :                     .unwrap(),
    1072            8 :             );
    1073            8 :         }
    1074              : 
    1075            8 :         if xinfo & pg_constants::XACT_XINFO_HAS_TWOPHASE != 0 {
    1076            0 :             xid = buf.get_u32_le();
    1077            0 :             tracing::debug!("XLOG_XACT_COMMIT-XACT_XINFO_HAS_TWOPHASE xid {}", xid);
    1078            8 :         }
    1079              : 
    1080            8 :         let origin_lsn = if xinfo & pg_constants::XACT_XINFO_HAS_ORIGIN != 0 {
    1081            0 :             Lsn(buf.get_u64_le())
    1082              :         } else {
    1083            8 :             Lsn::INVALID
    1084              :         };
    1085            8 :         XlXactParsedRecord {
    1086            8 :             xid,
    1087            8 :             info,
    1088            8 :             xact_time,
    1089            8 :             xinfo,
    1090            8 :             db_id,
    1091            8 :             ts_id,
    1092            8 :             subxacts,
    1093            8 :             xnodes,
    1094            8 :             origin_lsn,
    1095            8 :         }
    1096            8 :     }
    1097              : }
    1098              : 
    1099              : #[repr(C)]
    1100              : #[derive(Debug)]
    1101              : pub struct XlClogTruncate {
    1102              :     pub pageno: u32,
    1103              :     pub oldest_xid: TransactionId,
    1104              :     pub oldest_xid_db: Oid,
    1105              : }
    1106              : 
    1107              : impl XlClogTruncate {
    1108            0 :     pub fn decode(buf: &mut Bytes, pg_version: u32) -> XlClogTruncate {
    1109            0 :         XlClogTruncate {
    1110            0 :             pageno: if pg_version < 17 {
    1111            0 :                 buf.get_u32_le()
    1112              :             } else {
    1113            0 :                 buf.get_u64_le() as u32
    1114              :             },
    1115            0 :             oldest_xid: buf.get_u32_le(),
    1116            0 :             oldest_xid_db: buf.get_u32_le(),
    1117            0 :         }
    1118            0 :     }
    1119              : }
    1120              : 
    1121              : #[repr(C)]
    1122              : #[derive(Debug)]
    1123              : pub struct XlLogicalMessage {
    1124              :     pub db_id: Oid,
    1125              :     pub transactional: bool,
    1126              :     pub prefix_size: usize,
    1127              :     pub message_size: usize,
    1128              : }
    1129              : 
    1130              : impl XlLogicalMessage {
    1131            0 :     pub fn decode(buf: &mut Bytes) -> XlLogicalMessage {
    1132            0 :         XlLogicalMessage {
    1133            0 :             db_id: buf.get_u32_le(),
    1134            0 :             transactional: buf.get_u32_le() != 0, // 4-bytes alignment
    1135            0 :             prefix_size: buf.get_u64_le() as usize,
    1136            0 :             message_size: buf.get_u64_le() as usize,
    1137            0 :         }
    1138            0 :     }
    1139              : }
    1140              : 
    1141              : #[repr(C)]
    1142              : #[derive(Debug)]
    1143              : pub struct XlRunningXacts {
    1144              :     pub xcnt: u32,
    1145              :     pub subxcnt: u32,
    1146              :     pub subxid_overflow: bool,
    1147              :     pub next_xid: TransactionId,
    1148              :     pub oldest_running_xid: TransactionId,
    1149              :     pub latest_completed_xid: TransactionId,
    1150              :     pub xids: Vec<TransactionId>,
    1151              : }
    1152              : 
    1153              : impl XlRunningXacts {
    1154            0 :     pub fn decode(buf: &mut Bytes) -> XlRunningXacts {
    1155            0 :         let xcnt = buf.get_u32_le();
    1156            0 :         let subxcnt = buf.get_u32_le();
    1157            0 :         let subxid_overflow = buf.get_u32_le() != 0;
    1158            0 :         let next_xid = buf.get_u32_le();
    1159            0 :         let oldest_running_xid = buf.get_u32_le();
    1160            0 :         let latest_completed_xid = buf.get_u32_le();
    1161            0 :         let mut xids = Vec::new();
    1162            0 :         for _ in 0..(xcnt + subxcnt) {
    1163            0 :             xids.push(buf.get_u32_le());
    1164            0 :         }
    1165            0 :         XlRunningXacts {
    1166            0 :             xcnt,
    1167            0 :             subxcnt,
    1168            0 :             subxid_overflow,
    1169            0 :             next_xid,
    1170            0 :             oldest_running_xid,
    1171            0 :             latest_completed_xid,
    1172            0 :             xids,
    1173            0 :         }
    1174            0 :     }
    1175              : }
    1176              : 
    1177            0 : pub fn describe_postgres_wal_record(record: &Bytes) -> Result<String, DeserializeError> {
    1178            0 :     // TODO: It would be nice to use the PostgreSQL rmgrdesc infrastructure for this.
    1179            0 :     // Maybe use the postgres wal redo process, the same used for replaying WAL records?
    1180            0 :     // Or could we compile the rmgrdesc routines into the dump_layer_file() binary directly,
    1181            0 :     // without worrying about security?
    1182            0 :     //
    1183            0 :     // But for now, we have a hand-written code for a few common WAL record types here.
    1184            0 : 
    1185            0 :     let mut buf = record.clone();
    1186              : 
    1187              :     // 1. Parse XLogRecord struct
    1188              : 
    1189              :     // FIXME: assume little-endian here
    1190            0 :     let xlogrec = XLogRecord::from_bytes(&mut buf)?;
    1191              : 
    1192              :     let unknown_str: String;
    1193              : 
    1194            0 :     let result: &str = match xlogrec.xl_rmid {
    1195              :         pg_constants::RM_HEAP2_ID => {
    1196            0 :             let info = xlogrec.xl_info & pg_constants::XLOG_HEAP_OPMASK;
    1197            0 :             match info {
    1198            0 :                 pg_constants::XLOG_HEAP2_MULTI_INSERT => "HEAP2 MULTI_INSERT",
    1199            0 :                 pg_constants::XLOG_HEAP2_VISIBLE => "HEAP2 VISIBLE",
    1200              :                 _ => {
    1201            0 :                     unknown_str = format!("HEAP2 UNKNOWN_0x{:02x}", info);
    1202            0 :                     &unknown_str
    1203              :                 }
    1204              :             }
    1205              :         }
    1206              :         pg_constants::RM_HEAP_ID => {
    1207            0 :             let info = xlogrec.xl_info & pg_constants::XLOG_HEAP_OPMASK;
    1208            0 :             match info {
    1209            0 :                 pg_constants::XLOG_HEAP_INSERT => "HEAP INSERT",
    1210            0 :                 pg_constants::XLOG_HEAP_DELETE => "HEAP DELETE",
    1211            0 :                 pg_constants::XLOG_HEAP_UPDATE => "HEAP UPDATE",
    1212            0 :                 pg_constants::XLOG_HEAP_HOT_UPDATE => "HEAP HOT_UPDATE",
    1213              :                 _ => {
    1214            0 :                     unknown_str = format!("HEAP2 UNKNOWN_0x{:02x}", info);
    1215            0 :                     &unknown_str
    1216              :                 }
    1217              :             }
    1218              :         }
    1219              :         pg_constants::RM_XLOG_ID => {
    1220            0 :             let info = xlogrec.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
    1221            0 :             match info {
    1222            0 :                 pg_constants::XLOG_FPI => "XLOG FPI",
    1223            0 :                 pg_constants::XLOG_FPI_FOR_HINT => "XLOG FPI_FOR_HINT",
    1224              :                 _ => {
    1225            0 :                     unknown_str = format!("XLOG UNKNOWN_0x{:02x}", info);
    1226            0 :                     &unknown_str
    1227              :                 }
    1228              :             }
    1229              :         }
    1230            0 :         rmid => {
    1231            0 :             let info = xlogrec.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
    1232            0 : 
    1233            0 :             unknown_str = format!("UNKNOWN_RM_{} INFO_0x{:02x}", rmid, info);
    1234            0 :             &unknown_str
    1235              :         }
    1236              :     };
    1237              : 
    1238            0 :     Ok(String::from(result))
    1239            0 : }
        

Generated by: LCOV version 2.1-beta