LCOV - code coverage report
Current view: top level - libs/wal_decoder/src - decoder.rs (source / functions) Coverage Total Hit
Test: 1b0a6a0c05cee5a7de360813c8034804e105ce1c.info Lines: 31.7 % 789 250
Test Date: 2025-03-12 00:01:28 Functions: 58.8 % 17 10

            Line data    Source code
       1              : //! This module contains logic for decoding and interpreting
       2              : //! raw bytes which represent a raw Postgres WAL record.
       3              : 
       4              : use std::collections::HashMap;
       5              : 
       6              : use bytes::{Buf, Bytes};
       7              : use pageserver_api::key::rel_block_to_key;
       8              : use pageserver_api::reltag::{RelTag, SlruKind};
       9              : use pageserver_api::shard::ShardIdentity;
      10              : use postgres_ffi::pg_constants;
      11              : use postgres_ffi::relfile_utils::VISIBILITYMAP_FORKNUM;
      12              : use postgres_ffi::walrecord::*;
      13              : use utils::lsn::Lsn;
      14              : 
      15              : use crate::models::*;
      16              : use crate::serialized_batch::SerializedValueBatch;
      17              : 
      18              : impl InterpretedWalRecord {
      19              :     /// Decode and interpreted raw bytes which represent one Postgres WAL record.
      20              :     /// Data blocks which do not match any of the provided shard identities are filtered out.
      21              :     /// Shard 0 is a special case since it tracks all relation sizes. We only give it
      22              :     /// the keys that are being written as that is enough for updating relation sizes.
      23       292310 :     pub fn from_bytes_filtered(
      24       292310 :         buf: Bytes,
      25       292310 :         shards: &[ShardIdentity],
      26       292310 :         next_record_lsn: Lsn,
      27       292310 :         pg_version: u32,
      28       292310 :     ) -> anyhow::Result<HashMap<ShardIdentity, InterpretedWalRecord>> {
      29       292310 :         let mut decoded = DecodedWALRecord::default();
      30       292310 :         decode_wal_record(buf, &mut decoded, pg_version)?;
      31       292310 :         let xid = decoded.xl_xid;
      32              : 
      33       292310 :         let flush_uncommitted = if decoded.is_dbase_create_copy(pg_version) {
      34            0 :             FlushUncommittedRecords::Yes
      35              :         } else {
      36       292310 :             FlushUncommittedRecords::No
      37              :         };
      38              : 
      39       292310 :         let mut shard_records: HashMap<ShardIdentity, InterpretedWalRecord> =
      40       292310 :             HashMap::with_capacity(shards.len());
      41       584820 :         for shard in shards {
      42       292510 :             shard_records.insert(
      43       292510 :                 *shard,
      44       292510 :                 InterpretedWalRecord {
      45       292510 :                     metadata_record: None,
      46       292510 :                     batch: SerializedValueBatch::default(),
      47       292510 :                     next_record_lsn,
      48       292510 :                     flush_uncommitted,
      49       292510 :                     xid,
      50       292510 :                 },
      51       292510 :             );
      52       292510 :         }
      53              : 
      54       292310 :         MetadataRecord::from_decoded_filtered(
      55       292310 :             &decoded,
      56       292310 :             &mut shard_records,
      57       292310 :             next_record_lsn,
      58       292310 :             pg_version,
      59       292310 :         )?;
      60       292310 :         SerializedValueBatch::from_decoded_filtered(
      61       292310 :             decoded,
      62       292310 :             &mut shard_records,
      63       292310 :             next_record_lsn,
      64       292310 :             pg_version,
      65       292310 :         )?;
      66              : 
      67       292310 :         Ok(shard_records)
      68       292310 :     }
      69              : }
      70              : 
      71              : impl MetadataRecord {
      72              :     /// Populates the given `shard_records` with metadata records from this WAL record, if any,
      73              :     /// discarding those belonging to other shards.
      74              :     ///
      75              :     /// Only metadata records relevant for the given shards is emitted. Currently, most metadata
      76              :     /// records are broadcast to all shards for simplicity, but this should be improved.
      77       292310 :     fn from_decoded_filtered(
      78       292310 :         decoded: &DecodedWALRecord,
      79       292310 :         shard_records: &mut HashMap<ShardIdentity, InterpretedWalRecord>,
      80       292310 :         next_record_lsn: Lsn,
      81       292310 :         pg_version: u32,
      82       292310 :     ) -> anyhow::Result<()> {
      83       292310 :         // Note: this doesn't actually copy the bytes since
      84       292310 :         // the [`Bytes`] type implements it via a level of indirection.
      85       292310 :         let mut buf = decoded.record.clone();
      86       292310 :         buf.advance(decoded.main_data_offset);
      87              : 
      88              :         // First, generate metadata records from the decoded WAL record.
      89       292310 :         let metadata_record = match decoded.xl_rmid {
      90              :             pg_constants::RM_HEAP_ID | pg_constants::RM_HEAP2_ID => {
      91       290948 :                 Self::decode_heapam_record(&mut buf, decoded, pg_version)?
      92              :             }
      93            0 :             pg_constants::RM_NEON_ID => Self::decode_neonmgr_record(&mut buf, decoded, pg_version)?,
      94              :             // Handle other special record types
      95           32 :             pg_constants::RM_SMGR_ID => Self::decode_smgr_record(&mut buf, decoded)?,
      96            0 :             pg_constants::RM_DBASE_ID => Self::decode_dbase_record(&mut buf, decoded, pg_version)?,
      97              :             pg_constants::RM_TBLSPC_ID => {
      98            0 :                 tracing::trace!("XLOG_TBLSPC_CREATE/DROP is not handled yet");
      99            0 :                 None
     100              :             }
     101            0 :             pg_constants::RM_CLOG_ID => Self::decode_clog_record(&mut buf, decoded, pg_version)?,
     102              :             pg_constants::RM_XACT_ID => {
     103           48 :                 Self::decode_xact_record(&mut buf, decoded, next_record_lsn)?
     104              :             }
     105              :             pg_constants::RM_MULTIXACT_ID => {
     106            0 :                 Self::decode_multixact_record(&mut buf, decoded, pg_version)?
     107              :             }
     108            0 :             pg_constants::RM_RELMAP_ID => Self::decode_relmap_record(&mut buf, decoded)?,
     109              :             // This is an odd duck. It needs to go to all shards.
     110              :             // Since it uses the checkpoint image (that's initialized from CHECKPOINT_KEY
     111              :             // in WalIngest::new), we have to send the whole DecodedWalRecord::record to
     112              :             // the pageserver and decode it there.
     113              :             //
     114              :             // Alternatively, one can make the checkpoint part of the subscription protocol
     115              :             // to the pageserver. This should work fine, but can be done at a later point.
     116              :             pg_constants::RM_XLOG_ID => {
     117           60 :                 Self::decode_xlog_record(&mut buf, decoded, next_record_lsn)?
     118              :             }
     119              :             pg_constants::RM_LOGICALMSG_ID => {
     120          606 :                 Self::decode_logical_message_record(&mut buf, decoded)?
     121              :             }
     122           32 :             pg_constants::RM_STANDBY_ID => Self::decode_standby_record(&mut buf, decoded)?,
     123            0 :             pg_constants::RM_REPLORIGIN_ID => Self::decode_replorigin_record(&mut buf, decoded)?,
     124          584 :             _unexpected => {
     125          584 :                 // TODO: consider failing here instead of blindly doing something without
     126          584 :                 // understanding the protocol
     127          584 :                 None
     128              :             }
     129              :         };
     130              : 
     131              :         // Next, filter the metadata record by shard.
     132       292431 :         for (shard, record) in shard_records.iter_mut() {
     133          718 :             match metadata_record {
     134              :                 Some(
     135           24 :                     MetadataRecord::Heapam(HeapamRecord::ClearVmBits(ref clear_vm_bits))
     136            0 :                     | MetadataRecord::Neonrmgr(NeonrmgrRecord::ClearVmBits(ref clear_vm_bits)),
     137              :                 ) => {
     138              :                     // Route VM page updates to the shards that own them. VM pages are stored in the VM fork
     139              :                     // of the main relation. These are sharded and managed just like regular relation pages.
     140              :                     // See: https://github.com/neondatabase/neon/issues/9855
     141           24 :                     let is_local_vm_page = |heap_blk| {
     142           24 :                         let vm_blk = pg_constants::HEAPBLK_TO_MAPBLOCK(heap_blk);
     143           24 :                         shard.is_key_local(&rel_block_to_key(clear_vm_bits.vm_rel, vm_blk))
     144           24 :                     };
     145              :                     // Send the old and new VM page updates to their respective shards.
     146           24 :                     let updated_old_heap_blkno = clear_vm_bits
     147           24 :                         .old_heap_blkno
     148           24 :                         .filter(|&blkno| is_local_vm_page(blkno));
     149           24 :                     let updated_new_heap_blkno = clear_vm_bits
     150           24 :                         .new_heap_blkno
     151           24 :                         .filter(|&blkno| is_local_vm_page(blkno));
     152           24 :                     // If neither VM page belongs to this shard, discard the record.
     153           24 :                     if updated_old_heap_blkno.is_some() || updated_new_heap_blkno.is_some() {
     154              :                         // Clone the record and update it for the current shard.
     155           24 :                         let mut for_shard = metadata_record.clone();
     156           24 :                         match for_shard {
     157              :                             Some(
     158              :                                 MetadataRecord::Heapam(HeapamRecord::ClearVmBits(
     159           24 :                                     ref mut clear_vm_bits,
     160              :                                 ))
     161              :                                 | MetadataRecord::Neonrmgr(NeonrmgrRecord::ClearVmBits(
     162            0 :                                     ref mut clear_vm_bits,
     163              :                                 )),
     164           24 :                             ) => {
     165           24 :                                 clear_vm_bits.old_heap_blkno = updated_old_heap_blkno;
     166           24 :                                 clear_vm_bits.new_heap_blkno = updated_new_heap_blkno;
     167           24 :                                 record.metadata_record = for_shard;
     168           24 :                             }
     169              :                             _ => {
     170            0 :                                 unreachable!("for_shard is a clone of what we checked above")
     171              :                             }
     172              :                         }
     173            0 :                     }
     174              :                 }
     175              :                 Some(MetadataRecord::LogicalMessage(LogicalMessageRecord::Put(_))) => {
     176              :                     // Filter LogicalMessage records (AUX files) to only be stored on shard zero
     177          718 :                     if shard.is_shard_zero() {
     178          597 :                         record.metadata_record = metadata_record;
     179          597 :                         // No other shards should receive this record, so we stop traversing shards early.
     180          597 :                         break;
     181          121 :                     }
     182              :                 }
     183       291689 :                 _ => {
     184       291689 :                     // All other metadata records are sent to all shards.
     185       291689 :                     record.metadata_record = metadata_record.clone();
     186       291689 :                 }
     187              :             }
     188              :         }
     189              : 
     190       292310 :         Ok(())
     191       292310 :     }
     192              : 
     193       290948 :     fn decode_heapam_record(
     194       290948 :         buf: &mut Bytes,
     195       290948 :         decoded: &DecodedWALRecord,
     196       290948 :         pg_version: u32,
     197       290948 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     198       290948 :         // Handle VM bit updates that are implicitly part of heap records.
     199       290948 : 
     200       290948 :         // First, look at the record to determine which VM bits need
     201       290948 :         // to be cleared. If either of these variables is set, we
     202       290948 :         // need to clear the corresponding bits in the visibility map.
     203       290948 :         let mut new_heap_blkno: Option<u32> = None;
     204       290948 :         let mut old_heap_blkno: Option<u32> = None;
     205       290948 :         let mut flags = pg_constants::VISIBILITYMAP_VALID_BITS;
     206       290948 : 
     207       290948 :         match pg_version {
     208              :             14 => {
     209            0 :                 if decoded.xl_rmid == pg_constants::RM_HEAP_ID {
     210            0 :                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
     211            0 : 
     212            0 :                     if info == pg_constants::XLOG_HEAP_INSERT {
     213            0 :                         let xlrec = v14::XlHeapInsert::decode(buf);
     214            0 :                         assert_eq!(0, buf.remaining());
     215            0 :                         if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
     216            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     217            0 :                         }
     218            0 :                     } else if info == pg_constants::XLOG_HEAP_DELETE {
     219            0 :                         let xlrec = v14::XlHeapDelete::decode(buf);
     220            0 :                         if (xlrec.flags & pg_constants::XLH_DELETE_ALL_VISIBLE_CLEARED) != 0 {
     221            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     222            0 :                         }
     223            0 :                     } else if info == pg_constants::XLOG_HEAP_UPDATE
     224            0 :                         || info == pg_constants::XLOG_HEAP_HOT_UPDATE
     225              :                     {
     226            0 :                         let xlrec = v14::XlHeapUpdate::decode(buf);
     227            0 :                         // the size of tuple data is inferred from the size of the record.
     228            0 :                         // we can't validate the remaining number of bytes without parsing
     229            0 :                         // the tuple data.
     230            0 :                         if (xlrec.flags & pg_constants::XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED) != 0 {
     231            0 :                             old_heap_blkno = Some(decoded.blocks.last().unwrap().blkno);
     232            0 :                         }
     233            0 :                         if (xlrec.flags & pg_constants::XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED) != 0 {
     234            0 :                             // PostgreSQL only uses XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED on a
     235            0 :                             // non-HOT update where the new tuple goes to different page than
     236            0 :                             // the old one. Otherwise, only XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED is
     237            0 :                             // set.
     238            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     239            0 :                         }
     240            0 :                     } else if info == pg_constants::XLOG_HEAP_LOCK {
     241            0 :                         let xlrec = v14::XlHeapLock::decode(buf);
     242            0 :                         if (xlrec.flags & pg_constants::XLH_LOCK_ALL_FROZEN_CLEARED) != 0 {
     243            0 :                             old_heap_blkno = Some(decoded.blocks[0].blkno);
     244            0 :                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
     245            0 :                         }
     246            0 :                     }
     247            0 :                 } else if decoded.xl_rmid == pg_constants::RM_HEAP2_ID {
     248            0 :                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
     249            0 :                     if info == pg_constants::XLOG_HEAP2_MULTI_INSERT {
     250            0 :                         let xlrec = v14::XlHeapMultiInsert::decode(buf);
     251              : 
     252            0 :                         let offset_array_len =
     253            0 :                             if decoded.xl_info & pg_constants::XLOG_HEAP_INIT_PAGE > 0 {
     254              :                                 // the offsets array is omitted if XLOG_HEAP_INIT_PAGE is set
     255            0 :                                 0
     256              :                             } else {
     257            0 :                                 size_of::<u16>() * xlrec.ntuples as usize
     258              :                             };
     259            0 :                         assert_eq!(offset_array_len, buf.remaining());
     260              : 
     261            0 :                         if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
     262            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     263            0 :                         }
     264            0 :                     } else if info == pg_constants::XLOG_HEAP2_LOCK_UPDATED {
     265            0 :                         let xlrec = v14::XlHeapLockUpdated::decode(buf);
     266            0 :                         if (xlrec.flags & pg_constants::XLH_LOCK_ALL_FROZEN_CLEARED) != 0 {
     267            0 :                             old_heap_blkno = Some(decoded.blocks[0].blkno);
     268            0 :                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
     269            0 :                         }
     270            0 :                     }
     271              :                 } else {
     272            0 :                     anyhow::bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);
     273              :                 }
     274              :             }
     275              :             15 => {
     276       290948 :                 if decoded.xl_rmid == pg_constants::RM_HEAP_ID {
     277       290572 :                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
     278       290572 : 
     279       290572 :                     if info == pg_constants::XLOG_HEAP_INSERT {
     280       290552 :                         let xlrec = v15::XlHeapInsert::decode(buf);
     281       290552 :                         assert_eq!(0, buf.remaining());
     282       290552 :                         if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
     283            8 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     284       290544 :                         }
     285           20 :                     } else if info == pg_constants::XLOG_HEAP_DELETE {
     286            0 :                         let xlrec = v15::XlHeapDelete::decode(buf);
     287            0 :                         if (xlrec.flags & pg_constants::XLH_DELETE_ALL_VISIBLE_CLEARED) != 0 {
     288            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     289            0 :                         }
     290           20 :                     } else if info == pg_constants::XLOG_HEAP_UPDATE
     291            4 :                         || info == pg_constants::XLOG_HEAP_HOT_UPDATE
     292              :                     {
     293           16 :                         let xlrec = v15::XlHeapUpdate::decode(buf);
     294           16 :                         // the size of tuple data is inferred from the size of the record.
     295           16 :                         // we can't validate the remaining number of bytes without parsing
     296           16 :                         // the tuple data.
     297           16 :                         if (xlrec.flags & pg_constants::XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED) != 0 {
     298            0 :                             old_heap_blkno = Some(decoded.blocks.last().unwrap().blkno);
     299           16 :                         }
     300           16 :                         if (xlrec.flags & pg_constants::XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED) != 0 {
     301            0 :                             // PostgreSQL only uses XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED on a
     302            0 :                             // non-HOT update where the new tuple goes to different page than
     303            0 :                             // the old one. Otherwise, only XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED is
     304            0 :                             // set.
     305            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     306           16 :                         }
     307            4 :                     } else if info == pg_constants::XLOG_HEAP_LOCK {
     308            0 :                         let xlrec = v15::XlHeapLock::decode(buf);
     309            0 :                         if (xlrec.flags & pg_constants::XLH_LOCK_ALL_FROZEN_CLEARED) != 0 {
     310            0 :                             old_heap_blkno = Some(decoded.blocks[0].blkno);
     311            0 :                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
     312            0 :                         }
     313            4 :                     }
     314          376 :                 } else if decoded.xl_rmid == pg_constants::RM_HEAP2_ID {
     315          376 :                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
     316          376 :                     if info == pg_constants::XLOG_HEAP2_MULTI_INSERT {
     317           84 :                         let xlrec = v15::XlHeapMultiInsert::decode(buf);
     318              : 
     319           84 :                         let offset_array_len =
     320           84 :                             if decoded.xl_info & pg_constants::XLOG_HEAP_INIT_PAGE > 0 {
     321              :                                 // the offsets array is omitted if XLOG_HEAP_INIT_PAGE is set
     322            4 :                                 0
     323              :                             } else {
     324           80 :                                 size_of::<u16>() * xlrec.ntuples as usize
     325              :                             };
     326           84 :                         assert_eq!(offset_array_len, buf.remaining());
     327              : 
     328           84 :                         if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
     329           16 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     330           68 :                         }
     331          292 :                     } else if info == pg_constants::XLOG_HEAP2_LOCK_UPDATED {
     332            0 :                         let xlrec = v15::XlHeapLockUpdated::decode(buf);
     333            0 :                         if (xlrec.flags & pg_constants::XLH_LOCK_ALL_FROZEN_CLEARED) != 0 {
     334            0 :                             old_heap_blkno = Some(decoded.blocks[0].blkno);
     335            0 :                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
     336            0 :                         }
     337          292 :                     }
     338              :                 } else {
     339            0 :                     anyhow::bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);
     340              :                 }
     341              :             }
     342              :             16 => {
     343            0 :                 if decoded.xl_rmid == pg_constants::RM_HEAP_ID {
     344            0 :                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
     345            0 : 
     346            0 :                     if info == pg_constants::XLOG_HEAP_INSERT {
     347            0 :                         let xlrec = v16::XlHeapInsert::decode(buf);
     348            0 :                         assert_eq!(0, buf.remaining());
     349            0 :                         if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
     350            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     351            0 :                         }
     352            0 :                     } else if info == pg_constants::XLOG_HEAP_DELETE {
     353            0 :                         let xlrec = v16::XlHeapDelete::decode(buf);
     354            0 :                         if (xlrec.flags & pg_constants::XLH_DELETE_ALL_VISIBLE_CLEARED) != 0 {
     355            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     356            0 :                         }
     357            0 :                     } else if info == pg_constants::XLOG_HEAP_UPDATE
     358            0 :                         || info == pg_constants::XLOG_HEAP_HOT_UPDATE
     359              :                     {
     360            0 :                         let xlrec = v16::XlHeapUpdate::decode(buf);
     361            0 :                         // the size of tuple data is inferred from the size of the record.
     362            0 :                         // we can't validate the remaining number of bytes without parsing
     363            0 :                         // the tuple data.
     364            0 :                         if (xlrec.flags & pg_constants::XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED) != 0 {
     365            0 :                             old_heap_blkno = Some(decoded.blocks.last().unwrap().blkno);
     366            0 :                         }
     367            0 :                         if (xlrec.flags & pg_constants::XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED) != 0 {
     368            0 :                             // PostgreSQL only uses XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED on a
     369            0 :                             // non-HOT update where the new tuple goes to different page than
     370            0 :                             // the old one. Otherwise, only XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED is
     371            0 :                             // set.
     372            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     373            0 :                         }
     374            0 :                     } else if info == pg_constants::XLOG_HEAP_LOCK {
     375            0 :                         let xlrec = v16::XlHeapLock::decode(buf);
     376            0 :                         if (xlrec.flags & pg_constants::XLH_LOCK_ALL_FROZEN_CLEARED) != 0 {
     377            0 :                             old_heap_blkno = Some(decoded.blocks[0].blkno);
     378            0 :                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
     379            0 :                         }
     380            0 :                     }
     381            0 :                 } else if decoded.xl_rmid == pg_constants::RM_HEAP2_ID {
     382            0 :                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
     383            0 :                     if info == pg_constants::XLOG_HEAP2_MULTI_INSERT {
     384            0 :                         let xlrec = v16::XlHeapMultiInsert::decode(buf);
     385              : 
     386            0 :                         let offset_array_len =
     387            0 :                             if decoded.xl_info & pg_constants::XLOG_HEAP_INIT_PAGE > 0 {
     388              :                                 // the offsets array is omitted if XLOG_HEAP_INIT_PAGE is set
     389            0 :                                 0
     390              :                             } else {
     391            0 :                                 size_of::<u16>() * xlrec.ntuples as usize
     392              :                             };
     393            0 :                         assert_eq!(offset_array_len, buf.remaining());
     394              : 
     395            0 :                         if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
     396            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     397            0 :                         }
     398            0 :                     } else if info == pg_constants::XLOG_HEAP2_LOCK_UPDATED {
     399            0 :                         let xlrec = v16::XlHeapLockUpdated::decode(buf);
     400            0 :                         if (xlrec.flags & pg_constants::XLH_LOCK_ALL_FROZEN_CLEARED) != 0 {
     401            0 :                             old_heap_blkno = Some(decoded.blocks[0].blkno);
     402            0 :                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
     403            0 :                         }
     404            0 :                     }
     405              :                 } else {
     406            0 :                     anyhow::bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);
     407              :                 }
     408              :             }
     409              :             17 => {
     410            0 :                 if decoded.xl_rmid == pg_constants::RM_HEAP_ID {
     411            0 :                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
     412            0 : 
     413            0 :                     if info == pg_constants::XLOG_HEAP_INSERT {
     414            0 :                         let xlrec = v17::XlHeapInsert::decode(buf);
     415            0 :                         assert_eq!(0, buf.remaining());
     416            0 :                         if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
     417            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     418            0 :                         }
     419            0 :                     } else if info == pg_constants::XLOG_HEAP_DELETE {
     420            0 :                         let xlrec = v17::XlHeapDelete::decode(buf);
     421            0 :                         if (xlrec.flags & pg_constants::XLH_DELETE_ALL_VISIBLE_CLEARED) != 0 {
     422            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     423            0 :                         }
     424            0 :                     } else if info == pg_constants::XLOG_HEAP_UPDATE
     425            0 :                         || info == pg_constants::XLOG_HEAP_HOT_UPDATE
     426              :                     {
     427            0 :                         let xlrec = v17::XlHeapUpdate::decode(buf);
     428            0 :                         // the size of tuple data is inferred from the size of the record.
     429            0 :                         // we can't validate the remaining number of bytes without parsing
     430            0 :                         // the tuple data.
     431            0 :                         if (xlrec.flags & pg_constants::XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED) != 0 {
     432            0 :                             old_heap_blkno = Some(decoded.blocks.last().unwrap().blkno);
     433            0 :                         }
     434            0 :                         if (xlrec.flags & pg_constants::XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED) != 0 {
     435            0 :                             // PostgreSQL only uses XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED on a
     436            0 :                             // non-HOT update where the new tuple goes to different page than
     437            0 :                             // the old one. Otherwise, only XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED is
     438            0 :                             // set.
     439            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     440            0 :                         }
     441            0 :                     } else if info == pg_constants::XLOG_HEAP_LOCK {
     442            0 :                         let xlrec = v17::XlHeapLock::decode(buf);
     443            0 :                         if (xlrec.flags & pg_constants::XLH_LOCK_ALL_FROZEN_CLEARED) != 0 {
     444            0 :                             old_heap_blkno = Some(decoded.blocks[0].blkno);
     445            0 :                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
     446            0 :                         }
     447            0 :                     }
     448            0 :                 } else if decoded.xl_rmid == pg_constants::RM_HEAP2_ID {
     449            0 :                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
     450            0 :                     if info == pg_constants::XLOG_HEAP2_MULTI_INSERT {
     451            0 :                         let xlrec = v17::XlHeapMultiInsert::decode(buf);
     452              : 
     453            0 :                         let offset_array_len =
     454            0 :                             if decoded.xl_info & pg_constants::XLOG_HEAP_INIT_PAGE > 0 {
     455              :                                 // the offsets array is omitted if XLOG_HEAP_INIT_PAGE is set
     456            0 :                                 0
     457              :                             } else {
     458            0 :                                 size_of::<u16>() * xlrec.ntuples as usize
     459              :                             };
     460            0 :                         assert_eq!(offset_array_len, buf.remaining());
     461              : 
     462            0 :                         if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
     463            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     464            0 :                         }
     465            0 :                     } else if info == pg_constants::XLOG_HEAP2_LOCK_UPDATED {
     466            0 :                         let xlrec = v17::XlHeapLockUpdated::decode(buf);
     467            0 :                         if (xlrec.flags & pg_constants::XLH_LOCK_ALL_FROZEN_CLEARED) != 0 {
     468            0 :                             old_heap_blkno = Some(decoded.blocks[0].blkno);
     469            0 :                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
     470            0 :                         }
     471            0 :                     }
     472              :                 } else {
     473            0 :                     anyhow::bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);
     474              :                 }
     475              :             }
     476            0 :             _ => {}
     477              :         }
     478              : 
     479       290948 :         if new_heap_blkno.is_some() || old_heap_blkno.is_some() {
     480           24 :             let vm_rel = RelTag {
     481           24 :                 forknum: VISIBILITYMAP_FORKNUM,
     482           24 :                 spcnode: decoded.blocks[0].rnode_spcnode,
     483           24 :                 dbnode: decoded.blocks[0].rnode_dbnode,
     484           24 :                 relnode: decoded.blocks[0].rnode_relnode,
     485           24 :             };
     486           24 : 
     487           24 :             Ok(Some(MetadataRecord::Heapam(HeapamRecord::ClearVmBits(
     488           24 :                 ClearVmBits {
     489           24 :                     new_heap_blkno,
     490           24 :                     old_heap_blkno,
     491           24 :                     vm_rel,
     492           24 :                     flags,
     493           24 :                 },
     494           24 :             ))))
     495              :         } else {
     496       290924 :             Ok(None)
     497              :         }
     498       290948 :     }
     499              : 
     500            0 :     fn decode_neonmgr_record(
     501            0 :         buf: &mut Bytes,
     502            0 :         decoded: &DecodedWALRecord,
     503            0 :         pg_version: u32,
     504            0 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     505            0 :         // Handle VM bit updates that are implicitly part of heap records.
     506            0 : 
     507            0 :         // First, look at the record to determine which VM bits need
     508            0 :         // to be cleared. If either of these variables is set, we
     509            0 :         // need to clear the corresponding bits in the visibility map.
     510            0 :         let mut new_heap_blkno: Option<u32> = None;
     511            0 :         let mut old_heap_blkno: Option<u32> = None;
     512            0 :         let mut flags = pg_constants::VISIBILITYMAP_VALID_BITS;
     513            0 : 
     514            0 :         assert_eq!(decoded.xl_rmid, pg_constants::RM_NEON_ID);
     515              : 
     516            0 :         match pg_version {
     517              :             16 | 17 => {
     518            0 :                 let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
     519            0 : 
     520            0 :                 match info {
     521              :                     pg_constants::XLOG_NEON_HEAP_INSERT => {
     522            0 :                         let xlrec = v17::rm_neon::XlNeonHeapInsert::decode(buf);
     523            0 :                         assert_eq!(0, buf.remaining());
     524            0 :                         if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
     525            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     526            0 :                         }
     527              :                     }
     528              :                     pg_constants::XLOG_NEON_HEAP_DELETE => {
     529            0 :                         let xlrec = v17::rm_neon::XlNeonHeapDelete::decode(buf);
     530            0 :                         if (xlrec.flags & pg_constants::XLH_DELETE_ALL_VISIBLE_CLEARED) != 0 {
     531            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     532            0 :                         }
     533              :                     }
     534              :                     pg_constants::XLOG_NEON_HEAP_UPDATE
     535              :                     | pg_constants::XLOG_NEON_HEAP_HOT_UPDATE => {
     536            0 :                         let xlrec = v17::rm_neon::XlNeonHeapUpdate::decode(buf);
     537            0 :                         // the size of tuple data is inferred from the size of the record.
     538            0 :                         // we can't validate the remaining number of bytes without parsing
     539            0 :                         // the tuple data.
     540            0 :                         if (xlrec.flags & pg_constants::XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED) != 0 {
     541            0 :                             old_heap_blkno = Some(decoded.blocks.last().unwrap().blkno);
     542            0 :                         }
     543            0 :                         if (xlrec.flags & pg_constants::XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED) != 0 {
     544            0 :                             // PostgreSQL only uses XLH_UPDATE_NEW_ALL_VISIBLE_CLEARED on a
     545            0 :                             // non-HOT update where the new tuple goes to different page than
     546            0 :                             // the old one. Otherwise, only XLH_UPDATE_OLD_ALL_VISIBLE_CLEARED is
     547            0 :                             // set.
     548            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     549            0 :                         }
     550              :                     }
     551              :                     pg_constants::XLOG_NEON_HEAP_MULTI_INSERT => {
     552            0 :                         let xlrec = v17::rm_neon::XlNeonHeapMultiInsert::decode(buf);
     553              : 
     554            0 :                         let offset_array_len =
     555            0 :                             if decoded.xl_info & pg_constants::XLOG_HEAP_INIT_PAGE > 0 {
     556              :                                 // the offsets array is omitted if XLOG_HEAP_INIT_PAGE is set
     557            0 :                                 0
     558              :                             } else {
     559            0 :                                 size_of::<u16>() * xlrec.ntuples as usize
     560              :                             };
     561            0 :                         assert_eq!(offset_array_len, buf.remaining());
     562              : 
     563            0 :                         if (xlrec.flags & pg_constants::XLH_INSERT_ALL_VISIBLE_CLEARED) != 0 {
     564            0 :                             new_heap_blkno = Some(decoded.blocks[0].blkno);
     565            0 :                         }
     566              :                     }
     567              :                     pg_constants::XLOG_NEON_HEAP_LOCK => {
     568            0 :                         let xlrec = v17::rm_neon::XlNeonHeapLock::decode(buf);
     569            0 :                         if (xlrec.flags & pg_constants::XLH_LOCK_ALL_FROZEN_CLEARED) != 0 {
     570            0 :                             old_heap_blkno = Some(decoded.blocks[0].blkno);
     571            0 :                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
     572            0 :                         }
     573              :                     }
     574            0 :                     info => anyhow::bail!("Unknown WAL record type for Neon RMGR: {}", info),
     575              :                 }
     576              :             }
     577            0 :             _ => anyhow::bail!(
     578            0 :                 "Neon RMGR has no known compatibility with PostgreSQL version {}",
     579            0 :                 pg_version
     580            0 :             ),
     581              :         }
     582              : 
     583            0 :         if new_heap_blkno.is_some() || old_heap_blkno.is_some() {
     584            0 :             let vm_rel = RelTag {
     585            0 :                 forknum: VISIBILITYMAP_FORKNUM,
     586            0 :                 spcnode: decoded.blocks[0].rnode_spcnode,
     587            0 :                 dbnode: decoded.blocks[0].rnode_dbnode,
     588            0 :                 relnode: decoded.blocks[0].rnode_relnode,
     589            0 :             };
     590            0 : 
     591            0 :             Ok(Some(MetadataRecord::Neonrmgr(NeonrmgrRecord::ClearVmBits(
     592            0 :                 ClearVmBits {
     593            0 :                     new_heap_blkno,
     594            0 :                     old_heap_blkno,
     595            0 :                     vm_rel,
     596            0 :                     flags,
     597            0 :                 },
     598            0 :             ))))
     599              :         } else {
     600            0 :             Ok(None)
     601              :         }
     602            0 :     }
     603              : 
     604           32 :     fn decode_smgr_record(
     605           32 :         buf: &mut Bytes,
     606           32 :         decoded: &DecodedWALRecord,
     607           32 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     608           32 :         let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
     609           32 :         if info == pg_constants::XLOG_SMGR_CREATE {
     610           32 :             let create = XlSmgrCreate::decode(buf);
     611           32 :             let rel = RelTag {
     612           32 :                 spcnode: create.rnode.spcnode,
     613           32 :                 dbnode: create.rnode.dbnode,
     614           32 :                 relnode: create.rnode.relnode,
     615           32 :                 forknum: create.forknum,
     616           32 :             };
     617           32 : 
     618           32 :             return Ok(Some(MetadataRecord::Smgr(SmgrRecord::Create(SmgrCreate {
     619           32 :                 rel,
     620           32 :             }))));
     621            0 :         } else if info == pg_constants::XLOG_SMGR_TRUNCATE {
     622            0 :             let truncate = XlSmgrTruncate::decode(buf);
     623            0 :             return Ok(Some(MetadataRecord::Smgr(SmgrRecord::Truncate(truncate))));
     624            0 :         }
     625            0 : 
     626            0 :         Ok(None)
     627           32 :     }
     628              : 
     629            0 :     fn decode_dbase_record(
     630            0 :         buf: &mut Bytes,
     631            0 :         decoded: &DecodedWALRecord,
     632            0 :         pg_version: u32,
     633            0 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     634            0 :         // TODO: Refactor this to avoid the duplication between postgres versions.
     635            0 : 
     636            0 :         let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
     637            0 :         tracing::debug!(%info, %pg_version, "handle RM_DBASE_ID");
     638              : 
     639            0 :         if pg_version == 14 {
     640            0 :             if info == postgres_ffi::v14::bindings::XLOG_DBASE_CREATE {
     641            0 :                 let createdb = XlCreateDatabase::decode(buf);
     642            0 :                 tracing::debug!("XLOG_DBASE_CREATE v14");
     643              : 
     644            0 :                 let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
     645            0 :                     db_id: createdb.db_id,
     646            0 :                     tablespace_id: createdb.tablespace_id,
     647            0 :                     src_db_id: createdb.src_db_id,
     648            0 :                     src_tablespace_id: createdb.src_tablespace_id,
     649            0 :                 }));
     650            0 : 
     651            0 :                 return Ok(Some(record));
     652            0 :             } else if info == postgres_ffi::v14::bindings::XLOG_DBASE_DROP {
     653            0 :                 let dropdb = XlDropDatabase::decode(buf);
     654            0 : 
     655            0 :                 let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
     656            0 :                     db_id: dropdb.db_id,
     657            0 :                     tablespace_ids: dropdb.tablespace_ids,
     658            0 :                 }));
     659            0 : 
     660            0 :                 return Ok(Some(record));
     661            0 :             }
     662            0 :         } else if pg_version == 15 {
     663            0 :             if info == postgres_ffi::v15::bindings::XLOG_DBASE_CREATE_WAL_LOG {
     664            0 :                 tracing::debug!("XLOG_DBASE_CREATE_WAL_LOG: noop");
     665            0 :             } else if info == postgres_ffi::v15::bindings::XLOG_DBASE_CREATE_FILE_COPY {
     666              :                 // The XLOG record was renamed between v14 and v15,
     667              :                 // but the record format is the same.
     668              :                 // So we can reuse XlCreateDatabase here.
     669            0 :                 tracing::debug!("XLOG_DBASE_CREATE_FILE_COPY");
     670              : 
     671            0 :                 let createdb = XlCreateDatabase::decode(buf);
     672            0 :                 let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
     673            0 :                     db_id: createdb.db_id,
     674            0 :                     tablespace_id: createdb.tablespace_id,
     675            0 :                     src_db_id: createdb.src_db_id,
     676            0 :                     src_tablespace_id: createdb.src_tablespace_id,
     677            0 :                 }));
     678            0 : 
     679            0 :                 return Ok(Some(record));
     680            0 :             } else if info == postgres_ffi::v15::bindings::XLOG_DBASE_DROP {
     681            0 :                 let dropdb = XlDropDatabase::decode(buf);
     682            0 :                 let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
     683            0 :                     db_id: dropdb.db_id,
     684            0 :                     tablespace_ids: dropdb.tablespace_ids,
     685            0 :                 }));
     686            0 : 
     687            0 :                 return Ok(Some(record));
     688            0 :             }
     689            0 :         } else if pg_version == 16 {
     690            0 :             if info == postgres_ffi::v16::bindings::XLOG_DBASE_CREATE_WAL_LOG {
     691            0 :                 tracing::debug!("XLOG_DBASE_CREATE_WAL_LOG: noop");
     692            0 :             } else if info == postgres_ffi::v16::bindings::XLOG_DBASE_CREATE_FILE_COPY {
     693              :                 // The XLOG record was renamed between v14 and v15,
     694              :                 // but the record format is the same.
     695              :                 // So we can reuse XlCreateDatabase here.
     696            0 :                 tracing::debug!("XLOG_DBASE_CREATE_FILE_COPY");
     697              : 
     698            0 :                 let createdb = XlCreateDatabase::decode(buf);
     699            0 :                 let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
     700            0 :                     db_id: createdb.db_id,
     701            0 :                     tablespace_id: createdb.tablespace_id,
     702            0 :                     src_db_id: createdb.src_db_id,
     703            0 :                     src_tablespace_id: createdb.src_tablespace_id,
     704            0 :                 }));
     705            0 : 
     706            0 :                 return Ok(Some(record));
     707            0 :             } else if info == postgres_ffi::v16::bindings::XLOG_DBASE_DROP {
     708            0 :                 let dropdb = XlDropDatabase::decode(buf);
     709            0 :                 let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
     710            0 :                     db_id: dropdb.db_id,
     711            0 :                     tablespace_ids: dropdb.tablespace_ids,
     712            0 :                 }));
     713            0 : 
     714            0 :                 return Ok(Some(record));
     715            0 :             }
     716            0 :         } else if pg_version == 17 {
     717            0 :             if info == postgres_ffi::v17::bindings::XLOG_DBASE_CREATE_WAL_LOG {
     718            0 :                 tracing::debug!("XLOG_DBASE_CREATE_WAL_LOG: noop");
     719            0 :             } else if info == postgres_ffi::v17::bindings::XLOG_DBASE_CREATE_FILE_COPY {
     720              :                 // The XLOG record was renamed between v14 and v15,
     721              :                 // but the record format is the same.
     722              :                 // So we can reuse XlCreateDatabase here.
     723            0 :                 tracing::debug!("XLOG_DBASE_CREATE_FILE_COPY");
     724              : 
     725            0 :                 let createdb = XlCreateDatabase::decode(buf);
     726            0 :                 let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
     727            0 :                     db_id: createdb.db_id,
     728            0 :                     tablespace_id: createdb.tablespace_id,
     729            0 :                     src_db_id: createdb.src_db_id,
     730            0 :                     src_tablespace_id: createdb.src_tablespace_id,
     731            0 :                 }));
     732            0 : 
     733            0 :                 return Ok(Some(record));
     734            0 :             } else if info == postgres_ffi::v17::bindings::XLOG_DBASE_DROP {
     735            0 :                 let dropdb = XlDropDatabase::decode(buf);
     736            0 :                 let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
     737            0 :                     db_id: dropdb.db_id,
     738            0 :                     tablespace_ids: dropdb.tablespace_ids,
     739            0 :                 }));
     740            0 : 
     741            0 :                 return Ok(Some(record));
     742            0 :             }
     743            0 :         }
     744              : 
     745            0 :         Ok(None)
     746            0 :     }
     747              : 
     748            0 :     fn decode_clog_record(
     749            0 :         buf: &mut Bytes,
     750            0 :         decoded: &DecodedWALRecord,
     751            0 :         pg_version: u32,
     752            0 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     753            0 :         let info = decoded.xl_info & !pg_constants::XLR_INFO_MASK;
     754            0 : 
     755            0 :         if info == pg_constants::CLOG_ZEROPAGE {
     756            0 :             let pageno = if pg_version < 17 {
     757            0 :                 buf.get_u32_le()
     758              :             } else {
     759            0 :                 buf.get_u64_le() as u32
     760              :             };
     761            0 :             let segno = pageno / pg_constants::SLRU_PAGES_PER_SEGMENT;
     762            0 :             let rpageno = pageno % pg_constants::SLRU_PAGES_PER_SEGMENT;
     763            0 : 
     764            0 :             Ok(Some(MetadataRecord::Clog(ClogRecord::ZeroPage(
     765            0 :                 ClogZeroPage { segno, rpageno },
     766            0 :             ))))
     767              :         } else {
     768            0 :             assert!(info == pg_constants::CLOG_TRUNCATE);
     769            0 :             let xlrec = XlClogTruncate::decode(buf, pg_version);
     770            0 : 
     771            0 :             Ok(Some(MetadataRecord::Clog(ClogRecord::Truncate(
     772            0 :                 ClogTruncate {
     773            0 :                     pageno: xlrec.pageno,
     774            0 :                     oldest_xid: xlrec.oldest_xid,
     775            0 :                     oldest_xid_db: xlrec.oldest_xid_db,
     776            0 :                 },
     777            0 :             ))))
     778              :         }
     779            0 :     }
     780              : 
     781           48 :     fn decode_xact_record(
     782           48 :         buf: &mut Bytes,
     783           48 :         decoded: &DecodedWALRecord,
     784           48 :         lsn: Lsn,
     785           48 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     786           48 :         let info = decoded.xl_info & pg_constants::XLOG_XACT_OPMASK;
     787           48 :         let origin_id = decoded.origin_id;
     788           48 :         let xl_xid = decoded.xl_xid;
     789           48 : 
     790           48 :         if info == pg_constants::XLOG_XACT_COMMIT {
     791           16 :             let parsed = XlXactParsedRecord::decode(buf, decoded.xl_xid, decoded.xl_info);
     792           16 :             return Ok(Some(MetadataRecord::Xact(XactRecord::Commit(XactCommon {
     793           16 :                 parsed,
     794           16 :                 origin_id,
     795           16 :                 xl_xid,
     796           16 :                 lsn,
     797           16 :             }))));
     798           32 :         } else if info == pg_constants::XLOG_XACT_ABORT {
     799            0 :             let parsed = XlXactParsedRecord::decode(buf, decoded.xl_xid, decoded.xl_info);
     800            0 :             return Ok(Some(MetadataRecord::Xact(XactRecord::Abort(XactCommon {
     801            0 :                 parsed,
     802            0 :                 origin_id,
     803            0 :                 xl_xid,
     804            0 :                 lsn,
     805            0 :             }))));
     806           32 :         } else if info == pg_constants::XLOG_XACT_COMMIT_PREPARED {
     807            0 :             let parsed = XlXactParsedRecord::decode(buf, decoded.xl_xid, decoded.xl_info);
     808            0 :             return Ok(Some(MetadataRecord::Xact(XactRecord::CommitPrepared(
     809            0 :                 XactCommon {
     810            0 :                     parsed,
     811            0 :                     origin_id,
     812            0 :                     xl_xid,
     813            0 :                     lsn,
     814            0 :                 },
     815            0 :             ))));
     816           32 :         } else if info == pg_constants::XLOG_XACT_ABORT_PREPARED {
     817            0 :             let parsed = XlXactParsedRecord::decode(buf, decoded.xl_xid, decoded.xl_info);
     818            0 :             return Ok(Some(MetadataRecord::Xact(XactRecord::AbortPrepared(
     819            0 :                 XactCommon {
     820            0 :                     parsed,
     821            0 :                     origin_id,
     822            0 :                     xl_xid,
     823            0 :                     lsn,
     824            0 :                 },
     825            0 :             ))));
     826           32 :         } else if info == pg_constants::XLOG_XACT_PREPARE {
     827            0 :             return Ok(Some(MetadataRecord::Xact(XactRecord::Prepare(
     828            0 :                 XactPrepare {
     829            0 :                     xl_xid: decoded.xl_xid,
     830            0 :                     data: Bytes::copy_from_slice(&buf[..]),
     831            0 :                 },
     832            0 :             ))));
     833           32 :         }
     834           32 : 
     835           32 :         Ok(None)
     836           48 :     }
     837              : 
     838            0 :     fn decode_multixact_record(
     839            0 :         buf: &mut Bytes,
     840            0 :         decoded: &DecodedWALRecord,
     841            0 :         pg_version: u32,
     842            0 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     843            0 :         let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
     844            0 : 
     845            0 :         if info == pg_constants::XLOG_MULTIXACT_ZERO_OFF_PAGE
     846            0 :             || info == pg_constants::XLOG_MULTIXACT_ZERO_MEM_PAGE
     847              :         {
     848            0 :             let pageno = if pg_version < 17 {
     849            0 :                 buf.get_u32_le()
     850              :             } else {
     851            0 :                 buf.get_u64_le() as u32
     852              :             };
     853            0 :             let segno = pageno / pg_constants::SLRU_PAGES_PER_SEGMENT;
     854            0 :             let rpageno = pageno % pg_constants::SLRU_PAGES_PER_SEGMENT;
     855              : 
     856            0 :             let slru_kind = match info {
     857            0 :                 pg_constants::XLOG_MULTIXACT_ZERO_OFF_PAGE => SlruKind::MultiXactOffsets,
     858            0 :                 pg_constants::XLOG_MULTIXACT_ZERO_MEM_PAGE => SlruKind::MultiXactMembers,
     859            0 :                 _ => unreachable!(),
     860              :             };
     861              : 
     862            0 :             return Ok(Some(MetadataRecord::MultiXact(MultiXactRecord::ZeroPage(
     863            0 :                 MultiXactZeroPage {
     864            0 :                     slru_kind,
     865            0 :                     segno,
     866            0 :                     rpageno,
     867            0 :                 },
     868            0 :             ))));
     869            0 :         } else if info == pg_constants::XLOG_MULTIXACT_CREATE_ID {
     870            0 :             let xlrec = XlMultiXactCreate::decode(buf);
     871            0 :             return Ok(Some(MetadataRecord::MultiXact(MultiXactRecord::Create(
     872            0 :                 xlrec,
     873            0 :             ))));
     874            0 :         } else if info == pg_constants::XLOG_MULTIXACT_TRUNCATE_ID {
     875            0 :             let xlrec = XlMultiXactTruncate::decode(buf);
     876            0 :             return Ok(Some(MetadataRecord::MultiXact(MultiXactRecord::Truncate(
     877            0 :                 xlrec,
     878            0 :             ))));
     879            0 :         }
     880            0 : 
     881            0 :         Ok(None)
     882            0 :     }
     883              : 
     884            0 :     fn decode_relmap_record(
     885            0 :         buf: &mut Bytes,
     886            0 :         decoded: &DecodedWALRecord,
     887            0 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     888            0 :         let update = XlRelmapUpdate::decode(buf);
     889            0 : 
     890            0 :         let mut buf = decoded.record.clone();
     891            0 :         buf.advance(decoded.main_data_offset);
     892            0 :         // skip xl_relmap_update
     893            0 :         buf.advance(12);
     894            0 : 
     895            0 :         Ok(Some(MetadataRecord::Relmap(RelmapRecord::Update(
     896            0 :             RelmapUpdate {
     897            0 :                 update,
     898            0 :                 buf: Bytes::copy_from_slice(&buf[..]),
     899            0 :             },
     900            0 :         ))))
     901            0 :     }
     902              : 
     903           60 :     fn decode_xlog_record(
     904           60 :         buf: &mut Bytes,
     905           60 :         decoded: &DecodedWALRecord,
     906           60 :         lsn: Lsn,
     907           60 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     908           60 :         let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
     909           60 :         Ok(Some(MetadataRecord::Xlog(XlogRecord::Raw(RawXlogRecord {
     910           60 :             info,
     911           60 :             lsn,
     912           60 :             buf: buf.clone(),
     913           60 :         }))))
     914           60 :     }
     915              : 
     916          606 :     fn decode_logical_message_record(
     917          606 :         buf: &mut Bytes,
     918          606 :         decoded: &DecodedWALRecord,
     919          606 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     920          606 :         let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
     921          606 :         if info == pg_constants::XLOG_LOGICAL_MESSAGE {
     922          606 :             let xlrec = XlLogicalMessage::decode(buf);
     923          606 :             let prefix = std::str::from_utf8(&buf[0..xlrec.prefix_size - 1])?;
     924              : 
     925              :             #[cfg(feature = "testing")]
     926          606 :             if prefix == "neon-test" {
     927            0 :                 return Ok(Some(MetadataRecord::LogicalMessage(
     928            0 :                     LogicalMessageRecord::Failpoint,
     929            0 :                 )));
     930          606 :             }
     931              : 
     932          606 :             if let Some(path) = prefix.strip_prefix("neon-file:") {
     933          597 :                 let buf_size = xlrec.prefix_size + xlrec.message_size;
     934          597 :                 let buf = Bytes::copy_from_slice(&buf[xlrec.prefix_size..buf_size]);
     935          597 :                 return Ok(Some(MetadataRecord::LogicalMessage(
     936          597 :                     LogicalMessageRecord::Put(PutLogicalMessage {
     937          597 :                         path: path.to_string(),
     938          597 :                         buf,
     939          597 :                     }),
     940          597 :                 )));
     941            9 :             }
     942            0 :         }
     943              : 
     944            9 :         Ok(None)
     945          606 :     }
     946              : 
     947           32 :     fn decode_standby_record(
     948           32 :         buf: &mut Bytes,
     949           32 :         decoded: &DecodedWALRecord,
     950           32 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     951           32 :         let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
     952           32 :         if info == pg_constants::XLOG_RUNNING_XACTS {
     953            0 :             let xlrec = XlRunningXacts::decode(buf);
     954            0 :             return Ok(Some(MetadataRecord::Standby(StandbyRecord::RunningXacts(
     955            0 :                 StandbyRunningXacts {
     956            0 :                     oldest_running_xid: xlrec.oldest_running_xid,
     957            0 :                 },
     958            0 :             ))));
     959           32 :         }
     960           32 : 
     961           32 :         Ok(None)
     962           32 :     }
     963              : 
     964            0 :     fn decode_replorigin_record(
     965            0 :         buf: &mut Bytes,
     966            0 :         decoded: &DecodedWALRecord,
     967            0 :     ) -> anyhow::Result<Option<MetadataRecord>> {
     968            0 :         let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
     969            0 :         if info == pg_constants::XLOG_REPLORIGIN_SET {
     970            0 :             let xlrec = XlReploriginSet::decode(buf);
     971            0 :             return Ok(Some(MetadataRecord::Replorigin(ReploriginRecord::Set(
     972            0 :                 xlrec,
     973            0 :             ))));
     974            0 :         } else if info == pg_constants::XLOG_REPLORIGIN_DROP {
     975            0 :             let xlrec = XlReploriginDrop::decode(buf);
     976            0 :             return Ok(Some(MetadataRecord::Replorigin(ReploriginRecord::Drop(
     977            0 :                 xlrec,
     978            0 :             ))));
     979            0 :         }
     980            0 : 
     981            0 :         Ok(None)
     982            0 :     }
     983              : }
        

Generated by: LCOV version 2.1-beta