LCOV - code coverage report
Current view: top level - libs/postgres_ffi/src - lib.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 46.3 % 95 44
Test Date: 2025-03-12 16:10:49 Functions: 57.9 % 19 11

            Line data    Source code
       1              : #![allow(non_upper_case_globals)]
       2              : #![allow(non_camel_case_types)]
       3              : #![allow(non_snake_case)]
       4              : // bindgen creates some unsafe code with no doc comments.
       5              : #![allow(clippy::missing_safety_doc)]
       6              : // noted at 1.63 that in many cases there's u32 -> u32 transmutes in bindgen code.
       7              : #![allow(clippy::useless_transmute)]
       8              : // modules included with the postgres_ffi macro depend on the types of the specific version's
       9              : // types, and trigger a too eager lint.
      10              : #![allow(clippy::duplicate_mod)]
      11              : #![deny(clippy::undocumented_unsafe_blocks)]
      12              : 
      13              : use bytes::Bytes;
      14              : use utils::bin_ser::SerializeError;
      15              : use utils::lsn::Lsn;
      16              : 
      17              : macro_rules! postgres_ffi {
      18              :     ($version:ident) => {
      19              :         #[path = "."]
      20              :         pub mod $version {
      21              :             pub mod bindings {
      22              :                 // bindgen generates bindings for a lot of stuff we don't need
      23              :                 #![allow(dead_code)]
      24              :                 #![allow(unsafe_op_in_unsafe_fn)]
      25              :                 #![allow(clippy::undocumented_unsafe_blocks)]
      26              :                 #![allow(clippy::ptr_offset_with_cast)]
      27              : 
      28              :                 use serde::{Deserialize, Serialize};
      29              :                 include!(concat!(
      30              :                     env!("OUT_DIR"),
      31              :                     "/bindings_",
      32              :                     stringify!($version),
      33              :                     ".rs"
      34              :                 ));
      35              : 
      36              :                 include!(concat!("pg_constants_", stringify!($version), ".rs"));
      37              :             }
      38              :             pub mod controlfile_utils;
      39              :             pub mod nonrelfile_utils;
      40              :             pub mod wal_craft_test_export;
      41              :             pub mod wal_generator;
      42              :             pub mod waldecoder_handler;
      43              :             pub mod xlog_utils;
      44              : 
      45              :             pub const PG_MAJORVERSION: &str = stringify!($version);
      46              : 
      47              :             // Re-export some symbols from bindings
      48              :             pub use bindings::{CheckPoint, ControlFileData, DBState_DB_SHUTDOWNED, XLogRecord};
      49              : 
      50              :             pub const ZERO_CHECKPOINT: bytes::Bytes =
      51              :                 bytes::Bytes::from_static(&[0u8; xlog_utils::SIZEOF_CHECKPOINT]);
      52              :         }
      53              :     };
      54              : }
      55              : 
      56              : #[macro_export]
      57              : macro_rules! for_all_postgres_versions {
      58              :     ($macro:tt) => {
      59              :         $macro!(v14);
      60              :         $macro!(v15);
      61              :         $macro!(v16);
      62              :         $macro!(v17);
      63              :     };
      64              : }
      65              : 
      66              : for_all_postgres_versions! { postgres_ffi }
      67              : 
      68              : /// dispatch_pgversion
      69              : ///
      70              : /// Run a code block in a context where the postgres_ffi bindings for a
      71              : /// specific (supported) PostgreSQL version are `use`-ed in scope under the pgv
      72              : /// identifier.
      73              : /// If the provided pg_version is not supported, we panic!(), unless the
      74              : /// optional third argument was provided (in which case that code will provide
      75              : /// the default handling instead).
      76              : ///
      77              : /// Use like
      78              : ///
      79              : /// dispatch_pgversion!(my_pgversion, { pgv::constants::XLOG_DBASE_CREATE })
      80              : /// dispatch_pgversion!(my_pgversion, pgv::constants::XLOG_DBASE_CREATE)
      81              : ///
      82              : /// Other uses are for macro-internal purposes only and strictly unsupported.
      83              : ///
      84              : #[macro_export]
      85              : macro_rules! dispatch_pgversion {
      86              :     ($version:expr, $code:expr) => {
      87              :         dispatch_pgversion!($version, $code, panic!("Unknown PostgreSQL version {}", $version))
      88              :     };
      89              :     ($version:expr, $code:expr, $invalid_pgver_handling:expr) => {
      90              :         dispatch_pgversion!(
      91              :             $version => $code,
      92              :             default = $invalid_pgver_handling,
      93              :             pgversions = [
      94              :                 14 : v14,
      95              :                 15 : v15,
      96              :                 16 : v16,
      97              :                 17 : v17,
      98              :             ]
      99              :         )
     100              :     };
     101              :     ($pgversion:expr => $code:expr,
     102              :      default = $default:expr,
     103              :      pgversions = [$($sv:literal : $vsv:ident),+ $(,)?]) => {
     104              :         match ($pgversion) {
     105              :             $($sv => {
     106              :                 use $crate::$vsv as pgv;
     107              :                 $code
     108              :             },)+
     109              :             _ => {
     110              :                 $default
     111              :             }
     112              :         }
     113              :     };
     114              : }
     115              : 
     116              : #[macro_export]
     117              : macro_rules! enum_pgversion_dispatch {
     118              :     ($name:expr, $typ:ident, $bind:ident, $code:block) => {
     119              :         enum_pgversion_dispatch!(
     120              :             name = $name,
     121              :             bind = $bind,
     122              :             typ = $typ,
     123              :             code = $code,
     124              :             pgversions = [
     125              :                 V14 : v14,
     126              :                 V15 : v15,
     127              :                 V16 : v16,
     128              :                 V17 : v17,
     129              :             ]
     130              :         )
     131              :     };
     132              :     (name = $name:expr,
     133              :      bind = $bind:ident,
     134              :      typ = $typ:ident,
     135              :      code = $code:block,
     136              :      pgversions = [$($variant:ident : $md:ident),+ $(,)?]) => {
     137              :         match $name {
     138              :             $(
     139              :             self::$typ::$variant($bind) => {
     140              :                 use $crate::$md as pgv;
     141              :                 $code
     142              :             }
     143              :             ),+,
     144              :         }
     145              :     };
     146              : }
     147              : 
     148              : #[macro_export]
     149              : macro_rules! enum_pgversion {
     150              :     {$name:ident, pgv :: $t:ident} => {
     151              :         enum_pgversion!{
     152              :             name = $name,
     153              :             typ = $t,
     154              :             pgversions = [
     155              :                 V14 : v14,
     156              :                 V15 : v15,
     157              :                 V16 : v16,
     158              :                 V17 : v17,
     159              :             ]
     160              :         }
     161              :     };
     162              :     {$name:ident, pgv :: $p:ident :: $t:ident} => {
     163              :         enum_pgversion!{
     164              :             name = $name,
     165              :             path = $p,
     166              :             typ = $t,
     167              :             pgversions = [
     168              :                 V14 : v14,
     169              :                 V15 : v15,
     170              :                 V16 : v16,
     171              :                 V17 : v17,
     172              :             ]
     173              :         }
     174              :     };
     175              :     {name = $name:ident,
     176              :      typ = $t:ident,
     177              :      pgversions = [$($variant:ident : $md:ident),+ $(,)?]} => {
     178              :         pub enum $name {
     179              :             $($variant ( $crate::$md::$t )),+
     180              :         }
     181              :         impl self::$name {
     182            0 :             pub fn pg_version(&self) -> u32 {
     183            0 :                 enum_pgversion_dispatch!(self, $name, _ign, {
     184            0 :                     pgv::bindings::PG_MAJORVERSION_NUM
     185              :                 })
     186              :             }
     187              :         }
     188              :         $(
     189              :         impl Into<self::$name> for $crate::$md::$t {
     190           24 :             fn into(self) -> self::$name {
     191           24 :                 self::$name::$variant (self)
     192           24 :             }
     193              :         }
     194              :         )+
     195              :     };
     196              :     {name = $name:ident,
     197              :      path = $p:ident,
     198              :      typ = $t:ident,
     199              :      pgversions = [$($variant:ident : $md:ident),+ $(,)?]} => {
     200              :         pub enum $name {
     201              :             $($variant ($crate::$md::$p::$t)),+
     202              :         }
     203              :         impl $name {
     204              :             pub fn pg_version(&self) -> u32 {
     205              :                 enum_pgversion_dispatch!(self, $name, _ign, {
     206              :                     pgv::bindings::PG_MAJORVERSION_NUM
     207              :                 })
     208              :             }
     209              :         }
     210              :         $(
     211              :         impl Into<$name> for $crate::$md::$p::$t {
     212              :             fn into(self) -> $name {
     213              :                 $name::$variant (self)
     214              :             }
     215              :         }
     216              :         )+
     217              :     };
     218              : }
     219              : 
     220              : pub mod pg_constants;
     221              : pub mod relfile_utils;
     222              : pub mod walrecord;
     223              : 
     224              : // Export some widely used datatypes that are unlikely to change across Postgres versions
     225              : pub use v14::bindings::{
     226              :     BlockNumber, CheckPoint, ControlFileData, MultiXactId, OffsetNumber, Oid, PageHeaderData,
     227              :     RepOriginId, TimeLineID, TimestampTz, TransactionId, XLogRecPtr, XLogRecord, XLogSegNo, uint32,
     228              :     uint64,
     229              : };
     230              : // Likewise for these, although the assumption that these don't change is a little more iffy.
     231              : pub use v14::bindings::{MultiXactOffset, MultiXactStatus};
     232              : pub use v14::xlog_utils::{
     233              :     XLOG_SIZE_OF_XLOG_LONG_PHD, XLOG_SIZE_OF_XLOG_RECORD, XLOG_SIZE_OF_XLOG_SHORT_PHD,
     234              : };
     235              : 
     236              : // from pg_config.h. These can be changed with configure options --with-blocksize=BLOCKSIZE and
     237              : // --with-segsize=SEGSIZE, but assume the defaults for now.
     238              : pub const BLCKSZ: u16 = 8192;
     239              : pub const RELSEG_SIZE: u32 = 1024 * 1024 * 1024 / (BLCKSZ as u32);
     240              : pub const XLOG_BLCKSZ: usize = 8192;
     241              : pub const WAL_SEGMENT_SIZE: usize = 16 * 1024 * 1024;
     242              : 
     243              : pub const MAX_SEND_SIZE: usize = XLOG_BLCKSZ * 16;
     244              : 
     245              : // Export some version independent functions that are used outside of this mod
     246              : pub use v14::bindings::DBState_DB_SHUTDOWNED;
     247              : pub use v14::xlog_utils::{
     248              :     XLogFileName, encode_logical_message, get_current_timestamp, to_pg_timestamp,
     249              :     try_from_pg_timestamp,
     250              : };
     251              : 
     252          216 : pub fn bkpimage_is_compressed(bimg_info: u8, version: u32) -> bool {
     253          216 :     dispatch_pgversion!(version, pgv::bindings::bkpimg_is_compressed(bimg_info))
     254          216 : }
     255              : 
     256            5 : pub fn generate_wal_segment(
     257            5 :     segno: u64,
     258            5 :     system_id: u64,
     259            5 :     pg_version: u32,
     260            5 :     lsn: Lsn,
     261            5 : ) -> Result<Bytes, SerializeError> {
     262            5 :     assert_eq!(segno, lsn.segment_number(WAL_SEGMENT_SIZE));
     263              : 
     264              :     dispatch_pgversion!(
     265            5 :         pg_version,
     266            0 :         pgv::xlog_utils::generate_wal_segment(segno, system_id, lsn),
     267            0 :         Err(SerializeError::BadInput)
     268              :     )
     269            5 : }
     270              : 
     271            0 : pub fn generate_pg_control(
     272            0 :     pg_control_bytes: &[u8],
     273            0 :     checkpoint_bytes: &[u8],
     274            0 :     lsn: Lsn,
     275            0 :     pg_version: u32,
     276            0 : ) -> anyhow::Result<(Bytes, u64, bool)> {
     277            0 :     dispatch_pgversion!(
     278            0 :         pg_version,
     279            0 :         pgv::xlog_utils::generate_pg_control(pg_control_bytes, checkpoint_bytes, lsn),
     280            0 :         anyhow::bail!("Unknown version {}", pg_version)
     281              :     )
     282            0 : }
     283              : 
     284              : // PG timeline is always 1, changing it doesn't have any useful meaning in Neon.
     285              : //
     286              : // NOTE: this is not to be confused with Neon timelines; different concept!
     287              : //
     288              : // It's a shaky assumption, that it's always 1. We might import a
     289              : // PostgreSQL data directory that has gone through timeline bumps,
     290              : // for example. FIXME later.
     291              : pub const PG_TLI: u32 = 1;
     292              : 
     293              : //  See TransactionIdIsNormal in transam.h
     294            0 : pub const fn transaction_id_is_normal(id: TransactionId) -> bool {
     295            0 :     id > pg_constants::FIRST_NORMAL_TRANSACTION_ID
     296            0 : }
     297              : 
     298              : // See TransactionIdPrecedes in transam.c
     299            0 : pub const fn transaction_id_precedes(id1: TransactionId, id2: TransactionId) -> bool {
     300            0 :     /*
     301            0 :      * If either ID is a permanent XID then we can just do unsigned
     302            0 :      * comparison.  If both are normal, do a modulo-2^32 comparison.
     303            0 :      */
     304            0 : 
     305            0 :     if !(transaction_id_is_normal(id1)) || !transaction_id_is_normal(id2) {
     306            0 :         return id1 < id2;
     307            0 :     }
     308            0 : 
     309            0 :     let diff = id1.wrapping_sub(id2) as i32;
     310            0 :     diff < 0
     311            0 : }
     312              : 
     313              : // Check if page is not yet initialized (port of Postgres PageIsInit() macro)
     314           48 : pub fn page_is_new(pg: &[u8]) -> bool {
     315           48 :     pg[14] == 0 && pg[15] == 0 // pg_upper == 0
     316           48 : }
     317              : 
     318              : // ExtractLSN from page header
     319            0 : pub fn page_get_lsn(pg: &[u8]) -> Lsn {
     320            0 :     Lsn(
     321            0 :         ((u32::from_le_bytes(pg[0..4].try_into().unwrap()) as u64) << 32)
     322            0 :             | u32::from_le_bytes(pg[4..8].try_into().unwrap()) as u64,
     323            0 :     )
     324            0 : }
     325              : 
     326           36 : pub fn page_set_lsn(pg: &mut [u8], lsn: Lsn) {
     327           36 :     pg[0..4].copy_from_slice(&((lsn.0 >> 32) as u32).to_le_bytes());
     328           36 :     pg[4..8].copy_from_slice(&(lsn.0 as u32).to_le_bytes());
     329           36 : }
     330              : 
     331              : // This is port of function with the same name from freespace.c.
     332              : // The only difference is that it does not have "level" parameter because XLogRecordPageWithFreeSpace
     333              : // always call it with level=FSM_BOTTOM_LEVEL
     334            0 : pub fn fsm_logical_to_physical(addr: BlockNumber) -> BlockNumber {
     335            0 :     let mut leafno = addr;
     336              :     const FSM_TREE_DEPTH: u32 = if pg_constants::SLOTS_PER_FSM_PAGE >= 1626 {
     337              :         3
     338              :     } else {
     339              :         4
     340              :     };
     341              : 
     342              :     /* Count upper level nodes required to address the leaf page */
     343            0 :     let mut pages: BlockNumber = 0;
     344            0 :     for _l in 0..FSM_TREE_DEPTH {
     345            0 :         pages += leafno + 1;
     346            0 :         leafno /= pg_constants::SLOTS_PER_FSM_PAGE;
     347            0 :     }
     348              :     /* Turn the page count into 0-based block number */
     349            0 :     pages - 1
     350            0 : }
     351              : 
     352              : pub mod waldecoder {
     353              :     use std::num::NonZeroU32;
     354              : 
     355              :     use bytes::{Buf, Bytes, BytesMut};
     356              :     use thiserror::Error;
     357              :     use utils::lsn::Lsn;
     358              : 
     359              :     pub enum State {
     360              :         WaitingForRecord,
     361              :         ReassemblingRecord {
     362              :             recordbuf: BytesMut,
     363              :             contlen: NonZeroU32,
     364              :         },
     365              :         SkippingEverything {
     366              :             skip_until_lsn: Lsn,
     367              :         },
     368              :     }
     369              : 
     370              :     pub struct WalStreamDecoder {
     371              :         pub lsn: Lsn,
     372              :         pub pg_version: u32,
     373              :         pub inputbuf: BytesMut,
     374              :         pub state: State,
     375              :     }
     376              : 
     377              :     #[derive(Error, Debug, Clone)]
     378              :     #[error("{msg} at {lsn}")]
     379              :     pub struct WalDecodeError {
     380              :         pub msg: String,
     381              :         pub lsn: Lsn,
     382              :     }
     383              : 
     384              :     impl WalStreamDecoder {
     385         9801 :         pub fn new(lsn: Lsn, pg_version: u32) -> WalStreamDecoder {
     386         9801 :             WalStreamDecoder {
     387         9801 :                 lsn,
     388         9801 :                 pg_version,
     389         9801 :                 inputbuf: BytesMut::new(),
     390         9801 :                 state: State::WaitingForRecord,
     391         9801 :             }
     392         9801 :         }
     393              : 
     394              :         // The latest LSN position fed to the decoder.
     395         1364 :         pub fn available(&self) -> Lsn {
     396         1364 :             self.lsn + self.inputbuf.remaining() as u64
     397         1364 :         }
     398              : 
     399              :         /// Returns the LSN up to which the WAL decoder has processed.
     400              :         ///
     401              :         /// If [`Self::poll_decode`] returned a record, then this will return
     402              :         /// the end LSN of said record.
     403          606 :         pub fn lsn(&self) -> Lsn {
     404          606 :             self.lsn
     405          606 :         }
     406              : 
     407       967824 :         pub fn feed_bytes(&mut self, buf: &[u8]) {
     408       967824 :             self.inputbuf.extend_from_slice(buf);
     409       967824 :         }
     410              : 
     411      1284837 :         pub fn poll_decode(&mut self) -> Result<Option<(Lsn, Bytes)>, WalDecodeError> {
     412      1284837 :             dispatch_pgversion!(
     413      1284837 :                 self.pg_version,
     414              :                 {
     415              :                     use pgv::waldecoder_handler::WalStreamDecoderHandler;
     416         4167 :                     self.poll_decode_internal()
     417              :                 },
     418            0 :                 Err(WalDecodeError {
     419            0 :                     msg: format!("Unknown version {}", self.pg_version),
     420            0 :                     lsn: self.lsn,
     421            0 :                 })
     422              :             )
     423      1284837 :         }
     424              :     }
     425              : }
        

Generated by: LCOV version 2.1-beta