mirror of
https://github.com/neondatabase/neon.git
synced 2026-10-18 17:52:53 +00:00
pageserver: rename ingest functions
The goal of this commit is to make it clearer when we are ingesting the whole record versus when we are ingesting an action for the record. I also merged the VM bits clearing into one function since they were exactly the same.
This commit is contained in:
Vlad Lazar
+19
-107
@@ -323,7 +323,7 @@ impl WalIngest {
|
||||
if let Some(heapam_record) = maybe_heapam_record {
|
||||
match heapam_record {
|
||||
HeapamRecord::ClearVmBits(clear_vm_bits) => {
|
||||
self.ingest_heapam_record(clear_vm_bits, modification, ctx)
|
||||
self.ingest_clear_vm_bits(clear_vm_bits, modification, ctx)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
@@ -335,7 +335,7 @@ impl WalIngest {
|
||||
if let Some(neonrmgr_record) = maybe_nenonrmgr_record {
|
||||
match neonrmgr_record {
|
||||
NeonrmgrRecord::ClearVmBits(clear_vm_bits) => {
|
||||
self.ingest_neonrmgr_record(clear_vm_bits, modification, ctx)
|
||||
self.ingest_clear_vm_bits(clear_vm_bits, modification, ctx)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
@@ -386,11 +386,11 @@ impl WalIngest {
|
||||
|
||||
match clog_record {
|
||||
ClogRecord::ZeroPage(zero_page) => {
|
||||
self.ingest_clog_zero_page_record(zero_page, modification, ctx)
|
||||
self.ingest_clog_zero_page(zero_page, modification, ctx)
|
||||
.await?;
|
||||
}
|
||||
ClogRecord::Truncate(truncate) => {
|
||||
self.ingest_clog_truncate_record(truncate, modification, ctx)
|
||||
self.ingest_clog_truncate(truncate, modification, ctx)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
@@ -409,14 +409,14 @@ impl WalIngest {
|
||||
if let Some(multixact_record) = maybe_multixact_record {
|
||||
match multixact_record {
|
||||
MultiXactRecord::ZeroPage(zero_page) => {
|
||||
self.ingest_multixact_zero_page_record(zero_page, modification, ctx)
|
||||
self.ingest_multixact_zero_page(zero_page, modification, ctx)
|
||||
.await?;
|
||||
}
|
||||
MultiXactRecord::Create(create) => {
|
||||
self.ingest_multixact_create_record(modification, &create)?;
|
||||
self.ingest_multixact_create(modification, &create)?;
|
||||
}
|
||||
MultiXactRecord::Truncate(truncate) => {
|
||||
self.ingest_multixact_truncate_record(modification, &truncate, ctx)
|
||||
self.ingest_multixact_truncate(modification, &truncate, ctx)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
@@ -428,7 +428,7 @@ impl WalIngest {
|
||||
.unwrap();
|
||||
match relmap_record {
|
||||
RelmapRecord::Update(update) => {
|
||||
self.ingest_relmap_record(update, modification, ctx).await?;
|
||||
self.ingest_relmap_update(update, modification, ctx).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -446,7 +446,7 @@ impl WalIngest {
|
||||
|
||||
match xlog_record {
|
||||
XlogRecord::Raw(raw) => {
|
||||
self.ingest_xlog_record(raw, modification, ctx).await?;
|
||||
self.ingest_raw_xlog_record(raw, modification, ctx).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -456,7 +456,7 @@ impl WalIngest {
|
||||
if let Some(logical_message_record) = maybe_logical_message_record {
|
||||
match logical_message_record {
|
||||
LogicalMessageRecord::Put(put) => {
|
||||
self.ingest_logical_message_record(put, modification, ctx)
|
||||
self.ingest_logical_message_put(put, modification, ctx)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
@@ -635,7 +635,7 @@ impl WalIngest {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn ingest_heapam_record(
|
||||
async fn ingest_clear_vm_bits(
|
||||
&mut self,
|
||||
clear_vm_bits: ClearVmBits,
|
||||
modification: &mut DatadirModification<'_>,
|
||||
@@ -1028,94 +1028,6 @@ impl WalIngest {
|
||||
}
|
||||
}
|
||||
|
||||
async fn ingest_neonrmgr_record(
|
||||
&mut self,
|
||||
clear_vm_bits: ClearVmBits,
|
||||
modification: &mut DatadirModification<'_>,
|
||||
ctx: &RequestContext,
|
||||
) -> anyhow::Result<()> {
|
||||
let ClearVmBits {
|
||||
new_heap_blkno,
|
||||
old_heap_blkno,
|
||||
vm_rel,
|
||||
flags,
|
||||
} = clear_vm_bits;
|
||||
|
||||
let mut new_vm_blk = new_heap_blkno.map(pg_constants::HEAPBLK_TO_MAPBLOCK);
|
||||
let mut old_vm_blk = old_heap_blkno.map(pg_constants::HEAPBLK_TO_MAPBLOCK);
|
||||
|
||||
// Sometimes, Postgres seems to create heap WAL records with the
|
||||
// ALL_VISIBLE_CLEARED flag set, even though the bit in the VM page is
|
||||
// not set. In fact, it's possible that the VM page does not exist at all.
|
||||
// In that case, we don't want to store a record to clear the VM bit;
|
||||
// replaying it would fail to find the previous image of the page, because
|
||||
// it doesn't exist. So check if the VM page(s) exist, and skip the WAL
|
||||
// record if it doesn't.
|
||||
let vm_size = get_relsize(modification, vm_rel, ctx).await?;
|
||||
if let Some(blknum) = new_vm_blk {
|
||||
if blknum >= vm_size {
|
||||
new_vm_blk = None;
|
||||
}
|
||||
}
|
||||
if let Some(blknum) = old_vm_blk {
|
||||
if blknum >= vm_size {
|
||||
old_vm_blk = None;
|
||||
}
|
||||
}
|
||||
|
||||
if new_vm_blk.is_some() || old_vm_blk.is_some() {
|
||||
if new_vm_blk == old_vm_blk {
|
||||
// An UPDATE record that needs to clear the bits for both old and the
|
||||
// new page, both of which reside on the same VM page.
|
||||
self.put_rel_wal_record(
|
||||
modification,
|
||||
vm_rel,
|
||||
new_vm_blk.unwrap(),
|
||||
NeonWalRecord::ClearVisibilityMapFlags {
|
||||
new_heap_blkno,
|
||||
old_heap_blkno,
|
||||
flags,
|
||||
},
|
||||
ctx,
|
||||
)
|
||||
.await?;
|
||||
} else {
|
||||
// Clear VM bits for one heap page, or for two pages that reside on
|
||||
// different VM pages.
|
||||
if let Some(new_vm_blk) = new_vm_blk {
|
||||
self.put_rel_wal_record(
|
||||
modification,
|
||||
vm_rel,
|
||||
new_vm_blk,
|
||||
NeonWalRecord::ClearVisibilityMapFlags {
|
||||
new_heap_blkno,
|
||||
old_heap_blkno: None,
|
||||
flags,
|
||||
},
|
||||
ctx,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
if let Some(old_vm_blk) = old_vm_blk {
|
||||
self.put_rel_wal_record(
|
||||
modification,
|
||||
vm_rel,
|
||||
old_vm_blk,
|
||||
NeonWalRecord::ClearVisibilityMapFlags {
|
||||
new_heap_blkno: None,
|
||||
old_heap_blkno,
|
||||
flags,
|
||||
},
|
||||
ctx,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn decode_neonmgr_record(
|
||||
buf: &mut Bytes,
|
||||
decoded: &DecodedWALRecord,
|
||||
@@ -1808,7 +1720,7 @@ impl WalIngest {
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
async fn ingest_clog_truncate_record(
|
||||
async fn ingest_clog_truncate(
|
||||
&mut self,
|
||||
truncate: ClogTruncate,
|
||||
modification: &mut DatadirModification<'_>,
|
||||
@@ -1882,7 +1794,7 @@ impl WalIngest {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn ingest_clog_zero_page_record(
|
||||
async fn ingest_clog_zero_page(
|
||||
&mut self,
|
||||
zero_page: ClogZeroPage,
|
||||
modification: &mut DatadirModification<'_>,
|
||||
@@ -1930,7 +1842,7 @@ impl WalIngest {
|
||||
}
|
||||
}
|
||||
|
||||
fn ingest_multixact_create_record(
|
||||
fn ingest_multixact_create(
|
||||
&mut self,
|
||||
modification: &mut DatadirModification,
|
||||
xlrec: &XlMultiXactCreate,
|
||||
@@ -2032,7 +1944,7 @@ impl WalIngest {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn ingest_multixact_truncate_record(
|
||||
async fn ingest_multixact_truncate(
|
||||
&mut self,
|
||||
modification: &mut DatadirModification<'_>,
|
||||
xlrec: &XlMultiXactTruncate,
|
||||
@@ -2078,7 +1990,7 @@ impl WalIngest {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn ingest_multixact_zero_page_record(
|
||||
async fn ingest_multixact_zero_page(
|
||||
&mut self,
|
||||
zero_page: MultiXactZeroPage,
|
||||
modification: &mut DatadirModification<'_>,
|
||||
@@ -2140,7 +2052,7 @@ impl WalIngest {
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
async fn ingest_relmap_record(
|
||||
async fn ingest_relmap_update(
|
||||
&mut self,
|
||||
update: RelmapUpdate,
|
||||
modification: &mut DatadirModification<'_>,
|
||||
@@ -2171,7 +2083,7 @@ impl WalIngest {
|
||||
})))
|
||||
}
|
||||
|
||||
async fn ingest_xlog_record(
|
||||
async fn ingest_raw_xlog_record(
|
||||
&mut self,
|
||||
raw_record: RawXlogRecord,
|
||||
modification: &mut DatadirModification<'_>,
|
||||
@@ -2285,7 +2197,7 @@ impl WalIngest {
|
||||
})))
|
||||
}
|
||||
|
||||
async fn ingest_logical_message_record(
|
||||
async fn ingest_logical_message_put(
|
||||
&mut self,
|
||||
put: PutLogicalMessage,
|
||||
modification: &mut DatadirModification<'_>,
|
||||
|
||||
Reference in New Issue
Block a user