From 6fdba1a427a7e0072216c60a168f9b42cfdbc618 Mon Sep 17 00:00:00 2001 From: Vlad Lazar Date: Wed, 16 Oct 2024 18:36:27 +0200 Subject: [PATCH] pageserver: refactor clog records --- pageserver/src/walingest.rs | 119 +++++++++++++++++++++++++++--------- 1 file changed, 90 insertions(+), 29 deletions(-) diff --git a/pageserver/src/walingest.rs b/pageserver/src/walingest.rs index edc533c1e3..90c31aea93 100644 --- a/pageserver/src/walingest.rs +++ b/pageserver/src/walingest.rs @@ -152,6 +152,22 @@ struct DbaseDrop { tablespace_ids: Vec, } +enum ClogRecord { + ZeroPage(ClogZeroPage), + Truncate(ClogTruncate), +} + +struct ClogZeroPage { + segno: u32, + rpageno: u32, +} + +struct ClogTruncate { + pageno: u32, + oldest_xid: u32, + oldest_xid_db: u32, +} + impl WalIngest { pub async fn new( timeline: &Timeline, @@ -279,30 +295,21 @@ impl WalIngest { trace!("XLOG_TBLSPC_CREATE/DROP is not handled yet"); } pg_constants::RM_CLOG_ID => { - let info = decoded.xl_info & !pg_constants::XLR_INFO_MASK; + // [`Self::decode_clog_record`] may never fail and always returns. + // It has this interface to match all the other decoding methods. + let clog_record = Self::decode_clog_record(&mut buf, &decoded, pg_version) + .unwrap() + .unwrap(); - if info == pg_constants::CLOG_ZEROPAGE { - let pageno = if pg_version < 17 { - buf.get_u32_le() - } else { - buf.get_u64_le() as u32 - }; - let segno = pageno / pg_constants::SLRU_PAGES_PER_SEGMENT; - let rpageno = pageno % pg_constants::SLRU_PAGES_PER_SEGMENT; - self.put_slru_page_image( - modification, - SlruKind::Clog, - segno, - rpageno, - ZERO_PAGE.clone(), - ctx, - ) - .await?; - } else { - assert!(info == pg_constants::CLOG_TRUNCATE); - let xlrec = XlClogTruncate::decode(&mut buf, pg_version); - self.ingest_clog_truncate_record(modification, &xlrec, ctx) - .await?; + match clog_record { + ClogRecord::ZeroPage(zero_page) => { + self.ingest_clog_zero_page_record(zero_page, modification, ctx) + .await?; + } + ClogRecord::Truncate(truncate) => { + self.ingest_clog_truncate_record(truncate, modification, ctx) + .await?; + } } } pg_constants::RM_XACT_ID => { @@ -1783,13 +1790,19 @@ impl WalIngest { async fn ingest_clog_truncate_record( &mut self, + truncate: ClogTruncate, modification: &mut DatadirModification<'_>, - xlrec: &XlClogTruncate, ctx: &RequestContext, ) -> anyhow::Result<()> { + let ClogTruncate { + pageno, + oldest_xid, + oldest_xid_db, + } = truncate; + info!( "RM_CLOG_ID truncate pageno {} oldestXid {} oldestXidDB {}", - xlrec.pageno, xlrec.oldest_xid, xlrec.oldest_xid_db + pageno, oldest_xid, oldest_xid_db ); // In Postgres, oldestXid and oldestXidDB are updated in memory when the CLOG is @@ -1797,8 +1810,8 @@ impl WalIngest { // later. In Neon, a server can start at any LSN, not just on a checkpoint record, // so we keep the oldestXid and oldestXidDB up-to-date. enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, { - cp.oldestXid = xlrec.oldest_xid; - cp.oldestXidDB = xlrec.oldest_xid_db; + cp.oldestXid = oldest_xid; + cp.oldestXidDB = oldest_xid_db; }); self.checkpoint_modified = true; @@ -1815,7 +1828,7 @@ impl WalIngest { // the current endpoint page must not be eligible for removal. // See SimpleLruTruncate() in slru.c if dispatch_pgversion!(modification.tline.pg_version, { - pgv::nonrelfile_utils::clogpage_precedes(latest_page_number, xlrec.pageno) + pgv::nonrelfile_utils::clogpage_precedes(latest_page_number, pageno) }) { info!("could not truncate directory pg_xact apparent wraparound"); return Ok(()); @@ -1835,7 +1848,7 @@ impl WalIngest { let segpage = segno * pg_constants::SLRU_PAGES_PER_SEGMENT; let may_delete = dispatch_pgversion!(modification.tline.pg_version, { - pgv::nonrelfile_utils::slru_may_delete_clogsegment(segpage, xlrec.pageno) + pgv::nonrelfile_utils::slru_may_delete_clogsegment(segpage, pageno) }); if may_delete { @@ -1849,6 +1862,54 @@ impl WalIngest { Ok(()) } + async fn ingest_clog_zero_page_record( + &mut self, + zero_page: ClogZeroPage, + modification: &mut DatadirModification<'_>, + ctx: &RequestContext, + ) -> anyhow::Result<()> { + let ClogZeroPage { segno, rpageno } = zero_page; + + self.put_slru_page_image( + modification, + SlruKind::Clog, + segno, + rpageno, + ZERO_PAGE.clone(), + ctx, + ) + .await + } + + fn decode_clog_record( + buf: &mut Bytes, + decoded: &DecodedWALRecord, + pg_version: u32, + ) -> anyhow::Result> { + let info = decoded.xl_info & !pg_constants::XLR_INFO_MASK; + + if info == pg_constants::CLOG_ZEROPAGE { + let pageno = if pg_version < 17 { + buf.get_u32_le() + } else { + buf.get_u64_le() as u32 + }; + let segno = pageno / pg_constants::SLRU_PAGES_PER_SEGMENT; + let rpageno = pageno % pg_constants::SLRU_PAGES_PER_SEGMENT; + + Ok(Some(ClogRecord::ZeroPage(ClogZeroPage { segno, rpageno }))) + } else { + assert!(info == pg_constants::CLOG_TRUNCATE); + let xlrec = XlClogTruncate::decode(buf, pg_version); + + Ok(Some(ClogRecord::Truncate(ClogTruncate { + pageno: xlrec.pageno, + oldest_xid: xlrec.oldest_xid, + oldest_xid_db: xlrec.oldest_xid_db, + }))) + } + } + fn ingest_multixact_create_record( &mut self, modification: &mut DatadirModification,