LCOV - code coverage report
Current view: top level - libs/postgres_ffi/src - xlog_utils.rs (source / functions) Coverage Total Hit
Test: 32f4a56327bc9da697706839ed4836b2a00a408f.info Lines: 93.8 % 321 301
Test Date: 2024-02-07 07:37:29 Functions: 75.2 % 109 82

            Line data    Source code
       1              : //
       2              : // This file contains common utilities for dealing with PostgreSQL WAL files and
       3              : // LSNs.
       4              : //
       5              : // Many of these functions have been copied from PostgreSQL, and rewritten in
       6              : // Rust. That's why they don't follow the usual Rust naming conventions, they
       7              : // have been named the same as the corresponding PostgreSQL functions instead.
       8              : //
       9              : 
      10              : use crc32c::crc32c_append;
      11              : 
      12              : use super::super::waldecoder::WalStreamDecoder;
      13              : use super::bindings::{
      14              :     CheckPoint, ControlFileData, DBState_DB_SHUTDOWNED, FullTransactionId, TimeLineID, TimestampTz,
      15              :     XLogLongPageHeaderData, XLogPageHeaderData, XLogRecPtr, XLogRecord, XLogSegNo, XLOG_PAGE_MAGIC,
      16              : };
      17              : use super::PG_MAJORVERSION;
      18              : use crate::pg_constants;
      19              : use crate::PG_TLI;
      20              : use crate::{uint32, uint64, Oid};
      21              : use crate::{WAL_SEGMENT_SIZE, XLOG_BLCKSZ};
      22              : 
      23              : use bytes::BytesMut;
      24              : use bytes::{Buf, Bytes};
      25              : 
      26              : use log::*;
      27              : 
      28              : use serde::Serialize;
      29              : use std::fs::File;
      30              : use std::io::prelude::*;
      31              : use std::io::ErrorKind;
      32              : use std::io::SeekFrom;
      33              : use std::path::{Path, PathBuf};
      34              : use std::time::SystemTime;
      35              : use utils::bin_ser::DeserializeError;
      36              : use utils::bin_ser::SerializeError;
      37              : 
      38              : use utils::lsn::Lsn;
      39              : 
      40              : pub const XLOG_FNAME_LEN: usize = 24;
      41              : pub const XLP_FIRST_IS_CONTRECORD: u16 = 0x0001;
      42              : pub const XLP_REM_LEN_OFFS: usize = 2 + 2 + 4 + 8;
      43              : pub const XLOG_RECORD_CRC_OFFS: usize = 4 + 4 + 8 + 1 + 1 + 2;
      44              : 
      45              : pub const XLOG_SIZE_OF_XLOG_SHORT_PHD: usize = std::mem::size_of::<XLogPageHeaderData>();
      46              : pub const XLOG_SIZE_OF_XLOG_LONG_PHD: usize = std::mem::size_of::<XLogLongPageHeaderData>();
      47              : pub const XLOG_SIZE_OF_XLOG_RECORD: usize = std::mem::size_of::<XLogRecord>();
      48              : #[allow(clippy::identity_op)]
      49              : pub const SIZE_OF_XLOG_RECORD_DATA_HEADER_SHORT: usize = 1 * 2;
      50              : 
      51              : /// Interval of checkpointing metadata file. We should store metadata file to enforce
      52              : /// predicate that checkpoint.nextXid is larger than any XID in WAL.
      53              : /// But flushing checkpoint file for each transaction seems to be too expensive,
      54              : /// so XID_CHECKPOINT_INTERVAL is used to forward align nextXid and so perform
      55              : /// metadata checkpoint only once per XID_CHECKPOINT_INTERVAL transactions.
      56              : /// XID_CHECKPOINT_INTERVAL should not be larger than BLCKSZ*CLOG_XACTS_PER_BYTE
      57              : /// in order to let CLOG_TRUNCATE mechanism correctly extend CLOG.
      58              : const XID_CHECKPOINT_INTERVAL: u32 = 1024;
      59              : 
      60        12084 : pub fn XLogSegmentsPerXLogId(wal_segsz_bytes: usize) -> XLogSegNo {
      61        12084 :     (0x100000000u64 / wal_segsz_bytes as u64) as XLogSegNo
      62        12084 : }
      63              : 
      64          700 : pub fn XLogSegNoOffsetToRecPtr(
      65          700 :     segno: XLogSegNo,
      66          700 :     offset: u32,
      67          700 :     wal_segsz_bytes: usize,
      68          700 : ) -> XLogRecPtr {
      69          700 :     segno * (wal_segsz_bytes as u64) + (offset as u64)
      70          700 : }
      71              : 
      72         5882 : pub fn XLogFileName(tli: TimeLineID, logSegNo: XLogSegNo, wal_segsz_bytes: usize) -> String {
      73         5882 :     format!(
      74         5882 :         "{:>08X}{:>08X}{:>08X}",
      75         5882 :         tli,
      76         5882 :         logSegNo / XLogSegmentsPerXLogId(wal_segsz_bytes),
      77         5882 :         logSegNo % XLogSegmentsPerXLogId(wal_segsz_bytes)
      78         5882 :     )
      79         5882 : }
      80              : 
      81          320 : pub fn XLogFromFileName(fname: &str, wal_seg_size: usize) -> (XLogSegNo, TimeLineID) {
      82          320 :     let tli = u32::from_str_radix(&fname[0..8], 16).unwrap();
      83          320 :     let log = u32::from_str_radix(&fname[8..16], 16).unwrap() as XLogSegNo;
      84          320 :     let seg = u32::from_str_radix(&fname[16..24], 16).unwrap() as XLogSegNo;
      85          320 :     (log * XLogSegmentsPerXLogId(wal_seg_size) + seg, tli)
      86          320 : }
      87              : 
      88         1136 : pub fn IsXLogFileName(fname: &str) -> bool {
      89         8400 :     return fname.len() == XLOG_FNAME_LEN && fname.chars().all(|c| c.is_ascii_hexdigit());
      90         1136 : }
      91              : 
      92          726 : pub fn IsPartialXLogFileName(fname: &str) -> bool {
      93          726 :     fname.ends_with(".partial") && IsXLogFileName(&fname[0..fname.len() - 8])
      94          726 : }
      95              : 
      96              : /// If LSN points to the beginning of the page, then shift it to first record,
      97              : /// otherwise align on 8-bytes boundary (required for WAL records)
      98         1339 : pub fn normalize_lsn(lsn: Lsn, seg_sz: usize) -> Lsn {
      99         1339 :     if lsn.0 % XLOG_BLCKSZ as u64 == 0 {
     100           13 :         let hdr_size = if lsn.0 % seg_sz as u64 == 0 {
     101           10 :             XLOG_SIZE_OF_XLOG_LONG_PHD
     102              :         } else {
     103            3 :             XLOG_SIZE_OF_XLOG_SHORT_PHD
     104              :         };
     105           13 :         lsn + hdr_size as u64
     106              :     } else {
     107         1326 :         lsn.align()
     108              :     }
     109         1339 : }
     110              : 
     111          597 : pub fn generate_pg_control(
     112          597 :     pg_control_bytes: &[u8],
     113          597 :     checkpoint_bytes: &[u8],
     114          597 :     lsn: Lsn,
     115          597 : ) -> anyhow::Result<(Bytes, u64)> {
     116          597 :     let mut pg_control = ControlFileData::decode(pg_control_bytes)?;
     117          597 :     let mut checkpoint = CheckPoint::decode(checkpoint_bytes)?;
     118              : 
     119              :     // Generate new pg_control needed for bootstrap
     120          597 :     checkpoint.redo = normalize_lsn(lsn, WAL_SEGMENT_SIZE).0;
     121          597 : 
     122          597 :     //save new values in pg_control
     123          597 :     pg_control.checkPoint = 0;
     124          597 :     pg_control.checkPointCopy = checkpoint;
     125          597 :     pg_control.state = DBState_DB_SHUTDOWNED;
     126          597 : 
     127          597 :     Ok((pg_control.encode(), pg_control.system_identifier))
     128          597 : }
     129              : 
     130       821499 : pub fn get_current_timestamp() -> TimestampTz {
     131       821499 :     to_pg_timestamp(SystemTime::now())
     132       821499 : }
     133              : 
     134              : // Module to reduce the scope of the constants
     135              : mod timestamp_conversions {
     136              :     use std::time::Duration;
     137              : 
     138              :     use super::*;
     139              : 
     140              :     const UNIX_EPOCH_JDATE: u64 = 2440588; // == date2j(1970, 1, 1)
     141              :     const POSTGRES_EPOCH_JDATE: u64 = 2451545; // == date2j(2000, 1, 1)
     142              :     const SECS_PER_DAY: u64 = 86400;
     143              :     const USECS_PER_SEC: u64 = 1000000;
     144              :     const SECS_DIFF_UNIX_TO_POSTGRES_EPOCH: u64 =
     145              :         (POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY;
     146              : 
     147       821701 :     pub fn to_pg_timestamp(time: SystemTime) -> TimestampTz {
     148       821701 :         match time.duration_since(SystemTime::UNIX_EPOCH) {
     149       821701 :             Ok(n) => {
     150       821701 :                 ((n.as_secs() - SECS_DIFF_UNIX_TO_POSTGRES_EPOCH) * USECS_PER_SEC
     151       821701 :                     + n.subsec_micros() as u64) as i64
     152              :             }
     153            0 :             Err(_) => panic!("SystemTime before UNIX EPOCH!"),
     154              :         }
     155       821701 :     }
     156              : 
     157           22 :     pub fn from_pg_timestamp(time: TimestampTz) -> SystemTime {
     158           22 :         let time: u64 = time
     159           22 :             .try_into()
     160           22 :             .expect("timestamp before millenium (postgres epoch)");
     161           22 :         let since_unix_epoch = time + SECS_DIFF_UNIX_TO_POSTGRES_EPOCH * USECS_PER_SEC;
     162           22 :         SystemTime::UNIX_EPOCH
     163           22 :             .checked_add(Duration::from_micros(since_unix_epoch))
     164           22 :             .expect("SystemTime overflow")
     165           22 :     }
     166              : }
     167              : 
     168              : pub use timestamp_conversions::{from_pg_timestamp, to_pg_timestamp};
     169              : 
     170              : // Returns (aligned) end_lsn of the last record in data_dir with WAL segments.
     171              : // start_lsn must point to some previously known record boundary (beginning of
     172              : // the next record). If no valid record after is found, start_lsn is returned
     173              : // back.
     174          230 : pub fn find_end_of_wal(
     175          230 :     data_dir: &Path,
     176          230 :     wal_seg_size: usize,
     177          230 :     start_lsn: Lsn, // start reading WAL at this point; must point at record start_lsn.
     178          230 : ) -> anyhow::Result<Lsn> {
     179          230 :     let mut result = start_lsn;
     180          230 :     let mut curr_lsn = start_lsn;
     181          230 :     let mut buf = [0u8; XLOG_BLCKSZ];
     182          230 :     let pg_version = PG_MAJORVERSION[1..3].parse::<u32>().unwrap();
     183          230 :     debug!("find_end_of_wal PG_VERSION: {}", pg_version);
     184              : 
     185          230 :     let mut decoder = WalStreamDecoder::new(start_lsn, pg_version);
     186              : 
     187              :     // loop over segments
     188          258 :     loop {
     189          258 :         let segno = curr_lsn.segment_number(wal_seg_size);
     190          258 :         let seg_file_name = XLogFileName(PG_TLI, segno, wal_seg_size);
     191          258 :         let seg_file_path = data_dir.join(seg_file_name);
     192          258 :         match open_wal_segment(&seg_file_path)? {
     193              :             None => {
     194              :                 // no more segments
     195            6 :                 debug!(
     196            0 :                     "find_end_of_wal reached end at {:?}, segment {:?} doesn't exist",
     197              :                     result, seg_file_path
     198              :                 );
     199            6 :                 return Ok(result);
     200              :             }
     201          252 :             Some(mut segment) => {
     202          252 :                 let seg_offs = curr_lsn.segment_offset(wal_seg_size);
     203          252 :                 segment.seek(SeekFrom::Start(seg_offs as u64))?;
     204              :                 // loop inside segment
     205        54437 :                 while curr_lsn.segment_number(wal_seg_size) == segno {
     206        54409 :                     let bytes_read = segment.read(&mut buf)?;
     207        54409 :                     if bytes_read == 0 {
     208            0 :                         debug!(
     209            0 :                             "find_end_of_wal reached end at {:?}, EOF in segment {:?} at offset {}",
     210            0 :                             result,
     211            0 :                             seg_file_path,
     212            0 :                             curr_lsn.segment_offset(wal_seg_size)
     213              :                         );
     214            0 :                         return Ok(result);
     215        54409 :                     }
     216        54409 :                     curr_lsn += bytes_read as u64;
     217        54409 :                     decoder.feed_bytes(&buf[0..bytes_read]);
     218              : 
     219              :                     // advance result past all completely read records
     220              :                     loop {
     221      2673661 :                         match decoder.poll_decode() {
     222      2619252 :                             Ok(Some(record)) => result = record.0,
     223          224 :                             Err(e) => {
     224          224 :                                 debug!(
     225           42 :                                     "find_end_of_wal reached end at {:?}, decode error: {:?}",
     226              :                                     result, e
     227              :                                 );
     228          224 :                                 return Ok(result);
     229              :                             }
     230        54185 :                             Ok(None) => break, // need more data
     231              :                         }
     232              :                     }
     233              :                 }
     234              :             }
     235              :         }
     236              :     }
     237          230 : }
     238              : 
     239              : // Open .partial or full WAL segment file, if present.
     240          258 : fn open_wal_segment(seg_file_path: &Path) -> anyhow::Result<Option<File>> {
     241          258 :     let mut partial_path = seg_file_path.to_owned();
     242          258 :     partial_path.set_extension("partial");
     243          258 :     match File::open(partial_path) {
     244          224 :         Ok(file) => Ok(Some(file)),
     245           34 :         Err(e) => match e.kind() {
     246              :             ErrorKind::NotFound => {
     247              :                 // .partial not found, try full
     248           34 :                 match File::open(seg_file_path) {
     249           28 :                     Ok(file) => Ok(Some(file)),
     250            6 :                     Err(e) => match e.kind() {
     251            6 :                         ErrorKind::NotFound => Ok(None),
     252            0 :                         _ => Err(e.into()),
     253              :                     },
     254              :                 }
     255              :             }
     256            0 :             _ => Err(e.into()),
     257              :         },
     258              :     }
     259          258 : }
     260              : 
     261            0 : pub fn main() {
     262            0 :     let mut data_dir = PathBuf::new();
     263            0 :     data_dir.push(".");
     264            0 :     let wal_end = find_end_of_wal(&data_dir, WAL_SEGMENT_SIZE, Lsn(0)).unwrap();
     265            0 :     println!("wal_end={:?}", wal_end);
     266            0 : }
     267              : 
     268              : impl XLogRecord {
     269    158002741 :     pub fn from_slice(buf: &[u8]) -> Result<XLogRecord, DeserializeError> {
     270    158002741 :         use utils::bin_ser::LeSer;
     271    158002741 :         XLogRecord::des(buf)
     272    158002741 :     }
     273              : 
     274     73215430 :     pub fn from_bytes<B: Buf>(buf: &mut B) -> Result<XLogRecord, DeserializeError> {
     275     73215430 :         use utils::bin_ser::LeSer;
     276     73215430 :         XLogRecord::des_from(&mut buf.reader())
     277     73215430 :     }
     278              : 
     279           18 :     pub fn encode(&self) -> Result<Bytes, SerializeError> {
     280           18 :         use utils::bin_ser::LeSer;
     281           18 :         Ok(self.ser()?.into())
     282           18 :     }
     283              : 
     284              :     // Is this record an XLOG_SWITCH record? They need some special processing,
     285    158002740 :     pub fn is_xlog_switch_record(&self) -> bool {
     286    158002740 :         self.xl_info == pg_constants::XLOG_SWITCH && self.xl_rmid == pg_constants::RM_XLOG_ID
     287    158002740 :     }
     288              : }
     289              : 
     290              : impl XLogPageHeaderData {
     291      3305024 :     pub fn from_bytes<B: Buf>(buf: &mut B) -> Result<XLogPageHeaderData, DeserializeError> {
     292      3305024 :         use utils::bin_ser::LeSer;
     293      3305024 :         XLogPageHeaderData::des_from(&mut buf.reader())
     294      3305024 :     }
     295              : 
     296          599 :     pub fn encode(&self) -> Result<Bytes, SerializeError> {
     297          599 :         use utils::bin_ser::LeSer;
     298          599 :         self.ser().map(|b| b.into())
     299          599 :     }
     300              : }
     301              : 
     302              : impl XLogLongPageHeaderData {
     303         1585 :     pub fn from_bytes<B: Buf>(buf: &mut B) -> Result<XLogLongPageHeaderData, DeserializeError> {
     304         1585 :         use utils::bin_ser::LeSer;
     305         1585 :         XLogLongPageHeaderData::des_from(&mut buf.reader())
     306         1585 :     }
     307              : 
     308          604 :     pub fn encode(&self) -> Result<Bytes, SerializeError> {
     309          604 :         use utils::bin_ser::LeSer;
     310          604 :         self.ser().map(|b| b.into())
     311          604 :     }
     312              : }
     313              : 
     314              : pub const SIZEOF_CHECKPOINT: usize = std::mem::size_of::<CheckPoint>();
     315              : 
     316              : impl CheckPoint {
     317        33968 :     pub fn encode(&self) -> Result<Bytes, SerializeError> {
     318        33968 :         use utils::bin_ser::LeSer;
     319        33968 :         Ok(self.ser()?.into())
     320        33968 :     }
     321              : 
     322         2624 :     pub fn decode(buf: &[u8]) -> Result<CheckPoint, DeserializeError> {
     323         2624 :         use utils::bin_ser::LeSer;
     324         2624 :         CheckPoint::des(buf)
     325         2624 :     }
     326              : 
     327              :     /// Update next XID based on provided new_xid and stored epoch.
     328              :     /// Next XID should be greater than new_xid. This handles 32-bit
     329              :     /// XID wraparound correctly.
     330              :     ///
     331              :     /// Returns 'true' if the XID was updated.
     332     71973972 :     pub fn update_next_xid(&mut self, xid: u32) -> bool {
     333     71973972 :         // nextXid should be greater than any XID in WAL, so increment provided XID and check for wraparround.
     334     71973972 :         let mut new_xid = std::cmp::max(xid.wrapping_add(1), pg_constants::FIRST_NORMAL_TRANSACTION_ID);
     335     71973972 :         // To reduce number of metadata checkpoints, we forward align XID on XID_CHECKPOINT_INTERVAL.
     336     71973972 :         // XID_CHECKPOINT_INTERVAL should not be larger than BLCKSZ*CLOG_XACTS_PER_BYTE
     337     71973972 :         new_xid =
     338     71973972 :             new_xid.wrapping_add(XID_CHECKPOINT_INTERVAL - 1) & !(XID_CHECKPOINT_INTERVAL - 1);
     339     71973972 :         let full_xid = self.nextXid.value;
     340     71973972 :         let old_xid = full_xid as u32;
     341     71973972 :         if new_xid.wrapping_sub(old_xid) as i32 > 0 {
     342         7449 :             let mut epoch = full_xid >> 32;
     343         7449 :             if new_xid < old_xid {
     344            0 :                 // wrap-around
     345            0 :                 epoch += 1;
     346         7449 :             }
     347         7449 :             let nextXid = (epoch << 32) | new_xid as u64;
     348         7449 : 
     349         7449 :             if nextXid != self.nextXid.value {
     350         7449 :                 self.nextXid = FullTransactionId { value: nextXid };
     351         7449 :                 return true;
     352            0 :             }
     353     71966523 :         }
     354     71966523 :         false
     355     71973972 :     }
     356              : }
     357              : 
     358              : /// Generate new, empty WAL segment, with correct block headers at the first
     359              : /// page of the segment and the page that contains the given LSN.
     360              : /// We need this segment to start compute node.
     361          604 : pub fn generate_wal_segment(segno: u64, system_id: u64, lsn: Lsn) -> Result<Bytes, SerializeError> {
     362          604 :     let mut seg_buf = BytesMut::with_capacity(WAL_SEGMENT_SIZE);
     363          604 : 
     364          604 :     let pageaddr = XLogSegNoOffsetToRecPtr(segno, 0, WAL_SEGMENT_SIZE);
     365          604 : 
     366          604 :     let page_off = lsn.block_offset();
     367          604 :     let seg_off = lsn.segment_offset(WAL_SEGMENT_SIZE);
     368          604 : 
     369          604 :     let first_page_only = seg_off < XLOG_BLCKSZ;
     370          604 :     let (shdr_rem_len, infoflags) = if first_page_only {
     371            5 :         (seg_off, pg_constants::XLP_FIRST_IS_CONTRECORD)
     372              :     } else {
     373          599 :         (0, 0)
     374              :     };
     375              : 
     376          604 :     let hdr = XLogLongPageHeaderData {
     377          604 :         std: {
     378          604 :             XLogPageHeaderData {
     379          604 :                 xlp_magic: XLOG_PAGE_MAGIC as u16,
     380          604 :                 xlp_info: pg_constants::XLP_LONG_HEADER | infoflags,
     381          604 :                 xlp_tli: PG_TLI,
     382          604 :                 xlp_pageaddr: pageaddr,
     383          604 :                 xlp_rem_len: shdr_rem_len as u32,
     384          604 :                 ..Default::default() // Put 0 in padding fields.
     385          604 :             }
     386          604 :         },
     387          604 :         xlp_sysid: system_id,
     388          604 :         xlp_seg_size: WAL_SEGMENT_SIZE as u32,
     389          604 :         xlp_xlog_blcksz: XLOG_BLCKSZ as u32,
     390          604 :     };
     391              : 
     392          604 :     let hdr_bytes = hdr.encode()?;
     393          604 :     seg_buf.extend_from_slice(&hdr_bytes);
     394          604 : 
     395          604 :     //zero out the rest of the file
     396          604 :     seg_buf.resize(WAL_SEGMENT_SIZE, 0);
     397          604 : 
     398          604 :     if !first_page_only {
     399          599 :         let block_offset = lsn.page_offset_in_segment(WAL_SEGMENT_SIZE) as usize;
     400          599 :         let header = XLogPageHeaderData {
     401          599 :             xlp_magic: XLOG_PAGE_MAGIC as u16,
     402          599 :             xlp_info: if page_off >= pg_constants::SIZE_OF_PAGE_HEADER as u64 {
     403          597 :                 pg_constants::XLP_FIRST_IS_CONTRECORD
     404              :             } else {
     405            2 :                 0
     406              :             },
     407              :             xlp_tli: PG_TLI,
     408          599 :             xlp_pageaddr: lsn.page_lsn().0,
     409          599 :             xlp_rem_len: if page_off >= pg_constants::SIZE_OF_PAGE_HEADER as u64 {
     410          597 :                 page_off as u32
     411              :             } else {
     412            2 :                 0u32
     413              :             },
     414          599 :             ..Default::default() // Put 0 in padding fields.
     415              :         };
     416          599 :         let hdr_bytes = header.encode()?;
     417              : 
     418          599 :         debug_assert!(seg_buf.len() > block_offset + hdr_bytes.len());
     419          599 :         debug_assert_ne!(block_offset, 0);
     420              : 
     421          599 :         seg_buf[block_offset..block_offset + hdr_bytes.len()].copy_from_slice(&hdr_bytes[..]);
     422            5 :     }
     423              : 
     424          604 :     Ok(seg_buf.freeze())
     425          604 : }
     426              : 
     427              : #[repr(C)]
     428           18 : #[derive(Serialize)]
     429              : struct XlLogicalMessage {
     430              :     db_id: Oid,
     431              :     transactional: uint32, // bool, takes 4 bytes due to alignment in C structures
     432              :     prefix_size: uint64,
     433              :     message_size: uint64,
     434              : }
     435              : 
     436              : impl XlLogicalMessage {
     437            9 :     pub fn encode(&self) -> Bytes {
     438            9 :         use utils::bin_ser::LeSer;
     439            9 :         self.ser().unwrap().into()
     440            9 :     }
     441              : }
     442              : 
     443              : /// Create new WAL record for non-transactional logical message.
     444              : /// Used for creating artificial WAL for tests, as LogicalMessage
     445              : /// record is basically no-op.
     446              : ///
     447              : /// NOTE: This leaves the xl_prev field zero. The safekeeper and
     448              : /// pageserver tolerate that, but PostgreSQL does not.
     449            9 : pub fn encode_logical_message(prefix: &str, message: &str) -> Vec<u8> {
     450            9 :     let mut prefix_bytes: Vec<u8> = Vec::with_capacity(prefix.len() + 1);
     451            9 :     prefix_bytes.write_all(prefix.as_bytes()).unwrap();
     452            9 :     prefix_bytes.push(0);
     453            9 : 
     454            9 :     let message_bytes = message.as_bytes();
     455            9 : 
     456            9 :     let logical_message = XlLogicalMessage {
     457            9 :         db_id: 0,
     458            9 :         transactional: 0,
     459            9 :         prefix_size: prefix_bytes.len() as u64,
     460            9 :         message_size: message_bytes.len() as u64,
     461            9 :     };
     462            9 : 
     463            9 :     let mainrdata = logical_message.encode();
     464            9 :     let mainrdata_len: usize = mainrdata.len() + prefix_bytes.len() + message_bytes.len();
     465              :     // only short mainrdata is supported for now
     466            9 :     assert!(mainrdata_len <= 255);
     467            9 :     let mainrdata_len = mainrdata_len as u8;
     468            9 : 
     469            9 :     let mut data: Vec<u8> = vec![pg_constants::XLR_BLOCK_ID_DATA_SHORT, mainrdata_len];
     470            9 :     data.extend_from_slice(&mainrdata);
     471            9 :     data.extend_from_slice(&prefix_bytes);
     472            9 :     data.extend_from_slice(message_bytes);
     473            9 : 
     474            9 :     let total_len = XLOG_SIZE_OF_XLOG_RECORD + data.len();
     475            9 : 
     476            9 :     let mut header = XLogRecord {
     477            9 :         xl_tot_len: total_len as u32,
     478            9 :         xl_xid: 0,
     479            9 :         xl_prev: 0,
     480            9 :         xl_info: 0,
     481            9 :         xl_rmid: 21,
     482            9 :         __bindgen_padding_0: [0u8; 2usize],
     483            9 :         xl_crc: 0, // crc will be calculated later
     484            9 :     };
     485            9 : 
     486            9 :     let header_bytes = header.encode().expect("failed to encode header");
     487            9 :     let crc = crc32c_append(0, &data);
     488            9 :     let crc = crc32c_append(crc, &header_bytes[0..XLOG_RECORD_CRC_OFFS]);
     489            9 :     header.xl_crc = crc;
     490            9 : 
     491            9 :     let mut wal: Vec<u8> = Vec::new();
     492            9 :     wal.extend_from_slice(&header.encode().expect("failed to encode header"));
     493            9 :     wal.extend_from_slice(&data);
     494            9 : 
     495            9 :     // WAL start position must be aligned at 8 bytes,
     496            9 :     // this will add padding for the next WAL record.
     497            9 :     const PADDING: usize = 8;
     498            9 :     let padding_rem = wal.len() % PADDING;
     499            9 :     if padding_rem != 0 {
     500            0 :         wal.resize(wal.len() + PADDING - padding_rem, 0);
     501            9 :     }
     502              : 
     503            9 :     wal
     504            9 : }
     505              : 
     506              : #[cfg(test)]
     507              : mod tests {
     508              :     use super::*;
     509              : 
     510            6 :     #[test]
     511            6 :     fn test_ts_conversion() {
     512            6 :         let now = SystemTime::now();
     513            6 :         let round_trip = from_pg_timestamp(to_pg_timestamp(now));
     514            6 : 
     515            6 :         let now_since = now.duration_since(SystemTime::UNIX_EPOCH).unwrap();
     516            6 :         let round_trip_since = round_trip.duration_since(SystemTime::UNIX_EPOCH).unwrap();
     517            6 :         assert_eq!(now_since.as_micros(), round_trip_since.as_micros());
     518              : 
     519            6 :         let now_pg = get_current_timestamp();
     520            6 :         let round_trip_pg = to_pg_timestamp(from_pg_timestamp(now_pg));
     521            6 : 
     522            6 :         assert_eq!(now_pg, round_trip_pg);
     523            6 :     }
     524              : 
     525              :     // If you need to craft WAL and write tests for this module, put it at wal_craft crate.
     526              : }
        

Generated by: LCOV version 2.1-beta