mirror of
https://github.com/neondatabase/neon.git
synced 2026-10-12 23:03:04 +00:00
Fix dropping of relational/non-relational data from storage
This commit is contained in:
@@ -63,14 +63,17 @@ pub trait Timeline {
|
||||
///
|
||||
/// This will implicitly extend the relation, if the page is beyond the
|
||||
/// current end-of-file.
|
||||
fn put_wal_record(&self, tag: BufferTag, rec: WALRecord);
|
||||
fn put_wal_record(&self, tag: BufferTag, rec: WALRecord) -> Result<()>;
|
||||
|
||||
/// Like put_wal_record, but with ready-made image of the page.
|
||||
fn put_page_image(&self, tag: BufferTag, lsn: Lsn, img: Bytes);
|
||||
fn put_page_image(&self, tag: BufferTag, lsn: Lsn, img: Bytes) -> Result<()>;
|
||||
|
||||
/// Truncate relation
|
||||
fn put_truncation(&self, rel: RelTag, lsn: Lsn, nblocks: u32) -> Result<()>;
|
||||
|
||||
/// Drop relation or file segment
|
||||
fn put_drop(&self, tag: BufferTag, lsn: Lsn) -> Result<()>;
|
||||
|
||||
/// Put raw data
|
||||
fn put_raw_data(&self, key: RepositoryKey, data: &[u8]) -> Result<()>;
|
||||
|
||||
@@ -156,14 +159,18 @@ pub trait Timeline {
|
||||
blknum: blk.blkno,
|
||||
};
|
||||
|
||||
let rec = WALRecord {
|
||||
lsn,
|
||||
will_init: blk.will_init || blk.apply_image,
|
||||
rec: recdata.clone(),
|
||||
main_data_offset: decoded.main_data_offset as u32,
|
||||
};
|
||||
if blk.will_drop {
|
||||
self.put_drop(tag, lsn)?;
|
||||
} else {
|
||||
let rec = WALRecord {
|
||||
lsn,
|
||||
will_init: blk.will_init || blk.apply_image,
|
||||
rec: recdata.clone(),
|
||||
main_data_offset: decoded.main_data_offset as u32,
|
||||
};
|
||||
|
||||
self.put_wal_record(tag, rec);
|
||||
self.put_wal_record(tag, rec)?;
|
||||
}
|
||||
}
|
||||
|
||||
// Handle a few special record types
|
||||
@@ -285,6 +292,21 @@ pub trait Timeline {
|
||||
return Ok(dbs);
|
||||
}
|
||||
|
||||
fn get_tx_status(&self, xid: TransactionId, lsn: Lsn) -> Result<u8> {
|
||||
let tag = BufferTag {
|
||||
rel: RelTag {
|
||||
forknum: pg_constants::PG_XACT_FORKNUM,
|
||||
spcnode: 0,
|
||||
dbnode: 0,
|
||||
relnode: 0,
|
||||
},
|
||||
blknum: xid / pg_constants::CLOG_XACTS_PER_PAGE,
|
||||
};
|
||||
let clog_page = self.get_page_at_lsn(tag, lsn)?;
|
||||
let status = transaction_id_get_status(xid, &clog_page[..]);
|
||||
Ok(status)
|
||||
}
|
||||
|
||||
/// Get vector of prepared twophase transactions
|
||||
fn get_twophase(&self, lsn: Lsn) -> Result<Vec<TransactionId>> {
|
||||
let key = RepositoryKey {
|
||||
@@ -311,18 +333,7 @@ pub trait Timeline {
|
||||
}
|
||||
if key.lsn <= lsn {
|
||||
let xid = key.tag.blknum;
|
||||
let tag = BufferTag {
|
||||
rel: RelTag {
|
||||
forknum: pg_constants::PG_XACT_FORKNUM,
|
||||
spcnode: 0,
|
||||
dbnode: 0,
|
||||
relnode: 0,
|
||||
},
|
||||
blknum: xid / pg_constants::CLOG_XACTS_PER_PAGE,
|
||||
};
|
||||
let clog_page = self.get_page_at_lsn(tag, lsn)?;
|
||||
let status = transaction_id_get_status(xid, &clog_page[..]);
|
||||
if status == pg_constants::TRANSACTION_STATUS_IN_PROGRESS {
|
||||
if self.get_tx_status(xid, lsn)? == pg_constants::TRANSACTION_STATUS_IN_PROGRESS {
|
||||
gxacts.push(xid);
|
||||
}
|
||||
}
|
||||
@@ -567,11 +578,11 @@ mod tests {
|
||||
let tline = repo.create_empty_timeline(timelineid)?;
|
||||
|
||||
tline.init_valid_lsn(Lsn(1));
|
||||
tline.put_page_image(TEST_BUF(0), Lsn(2), TEST_IMG("foo blk 0 at 2"));
|
||||
tline.put_page_image(TEST_BUF(0), Lsn(2), TEST_IMG("foo blk 0 at 2"));
|
||||
tline.put_page_image(TEST_BUF(0), Lsn(3), TEST_IMG("foo blk 0 at 3"));
|
||||
tline.put_page_image(TEST_BUF(1), Lsn(4), TEST_IMG("foo blk 1 at 4"));
|
||||
tline.put_page_image(TEST_BUF(2), Lsn(5), TEST_IMG("foo blk 2 at 5"));
|
||||
tline.put_page_image(TEST_BUF(0), Lsn(2), TEST_IMG("foo blk 0 at 2"))?;
|
||||
tline.put_page_image(TEST_BUF(0), Lsn(2), TEST_IMG("foo blk 0 at 2"))?;
|
||||
tline.put_page_image(TEST_BUF(0), Lsn(3), TEST_IMG("foo blk 0 at 3"))?;
|
||||
tline.put_page_image(TEST_BUF(1), Lsn(4), TEST_IMG("foo blk 1 at 4"))?;
|
||||
tline.put_page_image(TEST_BUF(2), Lsn(5), TEST_IMG("foo blk 2 at 5"))?;
|
||||
|
||||
tline.advance_last_valid_lsn(Lsn(5));
|
||||
|
||||
@@ -661,7 +672,7 @@ mod tests {
|
||||
for i in 0..pg_constants::RELSEG_SIZE + 1 {
|
||||
let img = TEST_IMG(&format!("foo blk {} at {}", i, Lsn(lsn)));
|
||||
lsn += 1;
|
||||
tline.put_page_image(TEST_BUF(i as u32), Lsn(lsn), img);
|
||||
tline.put_page_image(TEST_BUF(i as u32), Lsn(lsn), img)?;
|
||||
}
|
||||
tline.advance_last_valid_lsn(Lsn(lsn));
|
||||
|
||||
|
||||
@@ -87,6 +87,7 @@ enum CacheEntryContent {
|
||||
PageImage(Bytes),
|
||||
WALRecord(WALRecord),
|
||||
Truncation,
|
||||
Drop,
|
||||
}
|
||||
|
||||
// The serialized representation of a CacheEntryContent begins with
|
||||
@@ -94,11 +95,12 @@ enum CacheEntryContent {
|
||||
// an UNUSED_VERSION_FLAG that is not represented in the CacheEntryContent
|
||||
// at all, you must peek into the first byte of the serialized representation
|
||||
// to read it.
|
||||
const CONTENT_PAGE_IMAGE: u8 = 1u8;
|
||||
const CONTENT_WAL_RECORD: u8 = 2u8;
|
||||
const CONTENT_TRUNCATION: u8 = 3u8;
|
||||
const CONTENT_PAGE_IMAGE: u8 = 0u8;
|
||||
const CONTENT_WAL_RECORD: u8 = 1u8;
|
||||
const CONTENT_TRUNCATION: u8 = 2u8;
|
||||
const CONTENT_DROP: u8 = 3u8;
|
||||
|
||||
const CONTENT_KIND_MASK: u8 = 3u8; // bitmask that covers the above
|
||||
const CONTENT_KIND_MASK: u8 = 3u8; // bitmask that covers the above
|
||||
|
||||
const UNUSED_VERSION_FLAG: u8 = 4u8;
|
||||
|
||||
@@ -117,6 +119,9 @@ impl CacheEntryContent {
|
||||
CacheEntryContent::Truncation => {
|
||||
buf.put_u8(CONTENT_TRUNCATION);
|
||||
}
|
||||
CacheEntryContent::Drop => {
|
||||
buf.put_u8(CONTENT_DROP);
|
||||
}
|
||||
}
|
||||
}
|
||||
pub fn unpack(buf: &mut Bytes) -> CacheEntryContent {
|
||||
@@ -131,6 +136,7 @@ impl CacheEntryContent {
|
||||
}
|
||||
CONTENT_WAL_RECORD => CacheEntryContent::WALRecord(WALRecord::unpack(buf)),
|
||||
CONTENT_TRUNCATION => CacheEntryContent::Truncation,
|
||||
CONTENT_DROP => CacheEntryContent::Drop,
|
||||
_ => unreachable!(),
|
||||
}
|
||||
}
|
||||
@@ -299,17 +305,17 @@ impl RocksTimeline {
|
||||
let mut base_img: Option<Bytes> = None;
|
||||
let mut records: Vec<WALRecord> = Vec::new();
|
||||
|
||||
let mut iter = self.db.raw_iterator();
|
||||
iter.seek_for_prev(key.to_bytes());
|
||||
let mut iter = self.iterator();
|
||||
iter.last(&key);
|
||||
|
||||
// Scan backwards, collecting the WAL records, until we hit an
|
||||
// old page image.
|
||||
while iter.valid() {
|
||||
let key = RepositoryKey::from_slice(iter.key().unwrap());
|
||||
let key = iter.key();
|
||||
if key.tag != tag {
|
||||
break;
|
||||
}
|
||||
let content = CacheEntryContent::from_slice(iter.value().unwrap());
|
||||
let content = CacheEntryContent::from_slice(iter.value());
|
||||
if let CacheEntryContent::PageImage(img) = content {
|
||||
// We have a base image. No need to dig deeper into the list of
|
||||
// records
|
||||
@@ -347,11 +353,11 @@ impl RocksTimeline {
|
||||
},
|
||||
lsn,
|
||||
};
|
||||
let mut iter = self.db.raw_iterator();
|
||||
let mut iter = self.iterator();
|
||||
loop {
|
||||
iter.seek_for_prev(key.to_bytes());
|
||||
iter.last(&key);
|
||||
if iter.valid() {
|
||||
let thiskey = RepositoryKey::from_slice(iter.key().unwrap());
|
||||
let thiskey = iter.key();
|
||||
if thiskey.tag.rel == rel {
|
||||
// Ignore entries with later LSNs.
|
||||
if thiskey.lsn > lsn {
|
||||
@@ -359,7 +365,7 @@ impl RocksTimeline {
|
||||
continue;
|
||||
}
|
||||
|
||||
let content = CacheEntryContent::from_slice(iter.value().unwrap());
|
||||
let content = CacheEntryContent::from_slice(iter.value());
|
||||
if let CacheEntryContent::Truncation = content {
|
||||
if thiskey.tag.blknum > 0 {
|
||||
key.tag.blknum = thiskey.tag.blknum - 1;
|
||||
@@ -378,6 +384,46 @@ impl RocksTimeline {
|
||||
Ok(0)
|
||||
}
|
||||
|
||||
///
|
||||
/// Drop relations with all its forks or non-relational file
|
||||
///
|
||||
fn drop(&self, tag: BufferTag) -> Result<()> {
|
||||
let mut iter = self.iterator();
|
||||
let mut key = RepositoryKey { tag, lsn: Lsn(u64::MAX) };
|
||||
if tag.rel.forknum == pg_constants::MAIN_FORKNUM {
|
||||
// if it is relation then remove all its blocks in all forks
|
||||
key.tag.blknum = u32::MAX;
|
||||
key.tag.rel.forknum = pg_constants::INIT_FORKNUM;
|
||||
} else {
|
||||
assert!(tag.rel.forknum > pg_constants::INIT_FORKNUM);
|
||||
}
|
||||
|
||||
iter.last(&key);
|
||||
while iter.valid() {
|
||||
let key = iter.key();
|
||||
if key.tag.rel.relnode != tag.rel.relnode ||
|
||||
key.tag.rel.spcnode != tag.rel.spcnode ||
|
||||
key.tag.rel.dbnode != tag.rel.dbnode ||
|
||||
(key.tag.rel.forknum != tag.rel.forknum &&
|
||||
tag.rel.forknum != pg_constants::MAIN_FORKNUM)
|
||||
{
|
||||
// no more entries belonging to this relation or file
|
||||
break;
|
||||
}
|
||||
let v = iter.value();
|
||||
if (v[0] & UNUSED_VERSION_FLAG) == 0 {
|
||||
let mut v = v.to_owned();
|
||||
v[0] |= UNUSED_VERSION_FLAG;
|
||||
self.db.put(key.to_bytes(), &v[..])?;
|
||||
} else {
|
||||
// already marked for deletion
|
||||
break;
|
||||
}
|
||||
iter.prev();
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn do_gc(&self, conf: &'static PageServerConf) -> Result<Bytes> {
|
||||
loop {
|
||||
thread::sleep(conf.gc_period);
|
||||
@@ -403,35 +449,50 @@ impl RocksTimeline {
|
||||
let mut inspected = 0u64;
|
||||
let mut deleted = 0u64;
|
||||
loop {
|
||||
let mut iter = self.db.raw_iterator();
|
||||
iter.seek_for_prev(maxkey.to_bytes());
|
||||
let mut iter = self.iterator();
|
||||
iter.last(&maxkey);
|
||||
if iter.valid() {
|
||||
let key = RepositoryKey::from_slice(iter.key().unwrap());
|
||||
let v = iter.value().unwrap();
|
||||
let key = iter.key();
|
||||
let v = iter.value();
|
||||
let flag = v[0];
|
||||
let last_lsn = key.lsn;
|
||||
|
||||
inspected += 1;
|
||||
|
||||
// Construct boundaries for old records cleanup
|
||||
maxkey.tag = key.tag;
|
||||
let last_lsn = key.lsn;
|
||||
maxkey.lsn = min(horizon, last_lsn); // do not remove last version
|
||||
|
||||
let mut minkey = maxkey.clone();
|
||||
minkey.lsn = Lsn(0); // first version
|
||||
|
||||
// Special handling of delete of PREPARE WAL record
|
||||
if (flag & CONTENT_KIND_MASK) == CONTENT_DROP {
|
||||
// If drop record in over the horizon then delete all entries from repository
|
||||
if last_lsn < horizon {
|
||||
self.drop(key.tag)?;
|
||||
}
|
||||
maxkey = minkey;
|
||||
continue;
|
||||
}
|
||||
|
||||
// Special handling of PREPARE transaction WAL record.
|
||||
if last_lsn < horizon
|
||||
&& key.tag.rel.forknum == pg_constants::PG_TWOPHASE_FORKNUM
|
||||
{
|
||||
if (v[0] & UNUSED_VERSION_FLAG) == 0 {
|
||||
let mut v = v.to_owned();
|
||||
v[0] |= UNUSED_VERSION_FLAG;
|
||||
self.db.put(key.to_bytes(), &v[..])?;
|
||||
deleted += 1;
|
||||
}
|
||||
maxkey = minkey;
|
||||
continue;
|
||||
}
|
||||
let xid = key.tag.blknum;
|
||||
// We can not remove information about uncompleted prepared transaction
|
||||
// even if it is over horizon
|
||||
if self.get_tx_status(xid, horizon)? != pg_constants::TRANSACTION_STATUS_IN_PROGRESS {
|
||||
if (v[0] & UNUSED_VERSION_FLAG) == 0 {
|
||||
let mut v = v.to_owned();
|
||||
v[0] |= UNUSED_VERSION_FLAG;
|
||||
self.db.put(key.to_bytes(), &v[..])?;
|
||||
deleted += 1;
|
||||
}
|
||||
}
|
||||
maxkey = minkey;
|
||||
continue;
|
||||
}
|
||||
// reconstruct most recent page version
|
||||
if (v[0] & CONTENT_KIND_MASK) == CONTENT_WAL_RECORD {
|
||||
// force reconstruction of most recent page version
|
||||
@@ -449,19 +510,19 @@ impl RocksTimeline {
|
||||
let new_img = self
|
||||
.walredo_mgr
|
||||
.request_redo(key.tag, key.lsn, base_img, records)?;
|
||||
self.put_page_image(key.tag, key.lsn, new_img.clone());
|
||||
self.put_page_image(key.tag, key.lsn, new_img.clone())?;
|
||||
|
||||
reconstructed += 1;
|
||||
}
|
||||
|
||||
iter.seek_for_prev(maxkey.to_bytes());
|
||||
iter.last(&maxkey);
|
||||
if iter.valid() {
|
||||
// do not remove last version
|
||||
if last_lsn > horizon {
|
||||
// locate most recent record before horizon
|
||||
let key = RepositoryKey::from_slice(iter.key().unwrap());
|
||||
let key = iter.key();
|
||||
if key.tag == maxkey.tag {
|
||||
let v = iter.value().unwrap();
|
||||
let v = iter.value();
|
||||
if (v[0] & CONTENT_KIND_MASK) == CONTENT_WAL_RECORD {
|
||||
let (base_img, records) =
|
||||
self.collect_records_for_apply(key.tag, key.lsn);
|
||||
@@ -470,7 +531,7 @@ impl RocksTimeline {
|
||||
let new_img = self
|
||||
.walredo_mgr
|
||||
.request_redo(key.tag, key.lsn, base_img, records)?;
|
||||
self.put_page_image(key.tag, key.lsn, new_img.clone());
|
||||
self.put_page_image(key.tag, key.lsn, new_img.clone())?;
|
||||
|
||||
truncated += 1;
|
||||
} else {
|
||||
@@ -497,11 +558,11 @@ impl RocksTimeline {
|
||||
if !iter.valid() {
|
||||
break;
|
||||
}
|
||||
let key = RepositoryKey::from_slice(iter.key().unwrap());
|
||||
let key = iter.key();
|
||||
if key.tag != maxkey.tag {
|
||||
break;
|
||||
}
|
||||
let v = iter.value().unwrap();
|
||||
let v = iter.value();
|
||||
if (v[0] & UNUSED_VERSION_FLAG) == 0 {
|
||||
let mut v = v.to_owned();
|
||||
v[0] |= UNUSED_VERSION_FLAG;
|
||||
@@ -547,13 +608,13 @@ impl Timeline for RocksTimeline {
|
||||
// ask the WAL redo service to reconstruct the page image from the WAL records.
|
||||
let key = RepositoryKey { tag, lsn };
|
||||
|
||||
let mut iter = self.db.raw_iterator();
|
||||
iter.seek_for_prev(key.to_bytes());
|
||||
let mut iter = self.iterator();
|
||||
iter.last(&key);
|
||||
|
||||
if iter.valid() {
|
||||
let key = RepositoryKey::from_slice(iter.key().unwrap());
|
||||
let key = iter.key();
|
||||
if key.tag == tag {
|
||||
let content = CacheEntryContent::from_slice(iter.value().unwrap());
|
||||
let content = CacheEntryContent::from_slice(iter.value());
|
||||
let page_img: Bytes;
|
||||
if let CacheEntryContent::PageImage(img) = content {
|
||||
page_img = img;
|
||||
@@ -562,7 +623,7 @@ impl Timeline for RocksTimeline {
|
||||
let (base_img, records) = self.collect_records_for_apply(tag, lsn);
|
||||
page_img = self.walredo_mgr.request_redo(tag, lsn, base_img, records)?;
|
||||
|
||||
self.put_page_image(tag, lsn, page_img.clone());
|
||||
self.put_page_image(tag, lsn, page_img.clone())?;
|
||||
} else {
|
||||
// No base image, and no WAL record. Huh?
|
||||
bail!(
|
||||
@@ -616,10 +677,10 @@ impl Timeline for RocksTimeline {
|
||||
},
|
||||
lsn,
|
||||
};
|
||||
let mut iter = self.db.raw_iterator();
|
||||
iter.seek_for_prev(key.to_bytes());
|
||||
let mut iter = self.iterator();
|
||||
iter.last(&key);
|
||||
if iter.valid() {
|
||||
let key = RepositoryKey::from_slice(iter.key().unwrap());
|
||||
let key = iter.key();
|
||||
if key.tag.rel == rel {
|
||||
debug!("Relation {} exists at {}", rel, lsn);
|
||||
return Ok(true);
|
||||
@@ -635,13 +696,13 @@ impl Timeline for RocksTimeline {
|
||||
///
|
||||
/// Adds a WAL record to the repository
|
||||
///
|
||||
fn put_wal_record(&self, tag: BufferTag, rec: WALRecord) {
|
||||
fn put_wal_record(&self, tag: BufferTag, rec: WALRecord) -> Result<()> {
|
||||
let lsn = rec.lsn;
|
||||
let key = RepositoryKey { tag, lsn };
|
||||
|
||||
let content = CacheEntryContent::WALRecord(rec);
|
||||
|
||||
let _res = self.db.put(key.to_bytes(), content.to_bytes());
|
||||
let _res = self.db.put(key.to_bytes(), content.to_bytes())?;
|
||||
trace!(
|
||||
"put_wal_record rel {} blk {} at {}",
|
||||
tag.rel,
|
||||
@@ -651,8 +712,20 @@ impl Timeline for RocksTimeline {
|
||||
|
||||
self.num_entries.fetch_add(1, Ordering::Relaxed);
|
||||
self.num_wal_records.fetch_add(1, Ordering::Relaxed);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
///
|
||||
/// Put drop record which should completely delete relation with all its forks
|
||||
/// or non-relational file from repository
|
||||
///
|
||||
fn put_drop(&self, tag: BufferTag, lsn: Lsn) -> Result<()> {
|
||||
let key = RepositoryKey { tag, lsn };
|
||||
let content = CacheEntryContent::Drop;
|
||||
let _res = self.db.put(key.to_bytes(), content.to_bytes())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
///
|
||||
/// Adds a relation-wide WAL record (like truncate) to the repository,
|
||||
/// associating it with all pages started with specified block number
|
||||
@@ -672,7 +745,7 @@ impl Timeline for RocksTimeline {
|
||||
lsn,
|
||||
};
|
||||
trace!("put_wal_record lsn: {}", key.lsn);
|
||||
let _res = self.db.put(key.to_bytes(), content.to_bytes());
|
||||
let _res = self.db.put(key.to_bytes(), content.to_bytes())?;
|
||||
}
|
||||
let n = (old_rel_size - nblocks) as u64;
|
||||
self.num_entries.fetch_add(n, Ordering::Relaxed);
|
||||
@@ -697,26 +770,12 @@ impl Timeline for RocksTimeline {
|
||||
///
|
||||
/// Memorize a full image of a page version
|
||||
///
|
||||
fn put_page_image(&self, tag: BufferTag, lsn: Lsn, img: Bytes) {
|
||||
let img_len = img.len();
|
||||
fn put_page_image(&self, tag: BufferTag, lsn: Lsn, img: Bytes) -> Result<()> {
|
||||
let key = RepositoryKey { tag, lsn };
|
||||
let content = CacheEntryContent::PageImage(img);
|
||||
|
||||
let mut val_buf = content.to_bytes();
|
||||
|
||||
// Zero size of page image indicates that page can be removed
|
||||
if img_len == 0 {
|
||||
if (val_buf[0] & UNUSED_VERSION_FLAG) != 0 {
|
||||
// records already marked for deletion
|
||||
return;
|
||||
} else {
|
||||
// delete truncated multixact page
|
||||
val_buf[0] |= UNUSED_VERSION_FLAG;
|
||||
}
|
||||
}
|
||||
|
||||
trace!("put_wal_record lsn: {}", key.lsn);
|
||||
let _res = self.db.put(key.to_bytes(), content.to_bytes());
|
||||
let _res = self.db.put(key.to_bytes(), content.to_bytes())?;
|
||||
|
||||
trace!(
|
||||
"put_page_image rel {} blk {} at {}",
|
||||
@@ -725,6 +784,7 @@ impl Timeline for RocksTimeline {
|
||||
lsn
|
||||
);
|
||||
self.num_page_images.fetch_add(1, Ordering::Relaxed);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn iterator(&self) -> Box<dyn RepositoryIterator + '_> {
|
||||
|
||||
@@ -279,7 +279,7 @@ fn restore_relfile(
|
||||
},
|
||||
blknum,
|
||||
};
|
||||
timeline.put_page_image(tag, lsn, Bytes::copy_from_slice(&buf));
|
||||
timeline.put_page_image(tag, lsn, Bytes::copy_from_slice(&buf))?;
|
||||
/*
|
||||
if oldest_lsn == 0 || p.lsn < oldest_lsn {
|
||||
oldest_lsn = p.lsn;
|
||||
@@ -335,7 +335,7 @@ fn restore_nonrel_file(
|
||||
},
|
||||
blknum,
|
||||
};
|
||||
timeline.put_page_image(tag, lsn, Bytes::copy_from_slice(&buffer[..]));
|
||||
timeline.put_page_image(tag, lsn, Bytes::copy_from_slice(&buffer[..]))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -369,7 +369,7 @@ fn restore_slru_file(
|
||||
},
|
||||
blknum,
|
||||
};
|
||||
timeline.put_page_image(tag, lsn, Bytes::copy_from_slice(&buf));
|
||||
timeline.put_page_image(tag, lsn, Bytes::copy_from_slice(&buf))?;
|
||||
/*
|
||||
if oldest_lsn == 0 || p.lsn < oldest_lsn {
|
||||
oldest_lsn = p.lsn;
|
||||
@@ -478,6 +478,6 @@ fn restore_wal(timeline: &dyn Timeline, timelineid: ZTimelineId, startpoint: Lsn
|
||||
}
|
||||
info!("reached end of WAL at {}", last_lsn);
|
||||
let checkpoint_bytes = encode_checkpoint(checkpoint);
|
||||
timeline.put_page_image(checkpoint_tag, Lsn(0), checkpoint_bytes);
|
||||
timeline.put_page_image(checkpoint_tag, Lsn(0), checkpoint_bytes)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -273,7 +273,8 @@ pub struct DecodedBkpBlock {
|
||||
/* Information on full-page image, if any */
|
||||
has_image: bool, /* has image, even for consistency checking */
|
||||
pub apply_image: bool, /* has image that should be restored */
|
||||
pub will_init: bool,
|
||||
pub will_init: bool, /* record intialize page content */
|
||||
pub will_drop: bool, /* record drops relation */
|
||||
//char *bkp_image;
|
||||
hole_offset: u16,
|
||||
hole_length: u16,
|
||||
@@ -298,6 +299,7 @@ impl DecodedBkpBlock {
|
||||
has_image: false,
|
||||
apply_image: false,
|
||||
will_init: false,
|
||||
will_drop: false,
|
||||
hole_offset: 0,
|
||||
hole_length: 0,
|
||||
bimg_len: 0,
|
||||
@@ -795,7 +797,13 @@ pub fn decode_wal_record(checkpoint: &mut CheckPoint, record: Bytes) -> DecodedW
|
||||
let mut blk = DecodedBkpBlock::new();
|
||||
blk.forknum = pg_constants::PG_XACT_FORKNUM;
|
||||
blk.blkno = buf.get_i32_le() as u32;
|
||||
blk.will_init = true;
|
||||
let info = xlogrec.xl_info & !pg_constants::XLR_INFO_MASK;
|
||||
if info == pg_constants::CLOG_ZEROPAGE {
|
||||
blk.will_init = true;
|
||||
} else {
|
||||
assert!(info == pg_constants::CLOG_TRUNCATE);
|
||||
blk.will_drop = true;
|
||||
}
|
||||
trace!("RM_CLOG_ID updates block {}", blk.blkno);
|
||||
blocks.push(blk);
|
||||
} else if xlogrec.xl_rmid == pg_constants::RM_XACT_ID {
|
||||
@@ -848,7 +856,13 @@ pub fn decode_wal_record(checkpoint: &mut CheckPoint, record: Bytes) -> DecodedW
|
||||
let spcnode = buf.get_u32_le();
|
||||
let dbnode = buf.get_u32_le();
|
||||
let relnode = buf.get_u32_le();
|
||||
//TODO handle this too?
|
||||
let mut blk = DecodedBkpBlock::new();
|
||||
blk.forknum = pg_constants::MAIN_FORKNUM;
|
||||
blk.rnode_spcnode = spcnode;
|
||||
blk.rnode_dbnode = dbnode;
|
||||
blk.rnode_relnode = relnode;
|
||||
blk.will_drop = true;
|
||||
blocks.push(blk);
|
||||
trace!(
|
||||
"XLOG_XACT_COMMIT relfilenode {}/{}/{}",
|
||||
spcnode,
|
||||
@@ -921,7 +935,13 @@ pub fn decode_wal_record(checkpoint: &mut CheckPoint, record: Bytes) -> DecodedW
|
||||
let spcnode = buf.get_u32_le();
|
||||
let dbnode = buf.get_u32_le();
|
||||
let relnode = buf.get_u32_le();
|
||||
//TODO save these too
|
||||
let mut blk = DecodedBkpBlock::new();
|
||||
blk.forknum = pg_constants::MAIN_FORKNUM;
|
||||
blk.rnode_spcnode = spcnode;
|
||||
blk.rnode_dbnode = dbnode;
|
||||
blk.rnode_relnode = relnode;
|
||||
blk.will_drop = true;
|
||||
blocks.push(blk);
|
||||
trace!(
|
||||
"XLOG_XACT_ABORT relfilenode {}/{}/{}",
|
||||
spcnode,
|
||||
@@ -1111,7 +1131,7 @@ pub fn decode_wal_record(checkpoint: &mut CheckPoint, record: Bytes) -> DecodedW
|
||||
let mut blk = DecodedBkpBlock::new();
|
||||
blk.forknum = pg_constants::PG_MXACT_OFFSETS_FORKNUM;
|
||||
blk.blkno = blkno;
|
||||
blk.will_init = true;
|
||||
blk.will_drop = true;
|
||||
blocks.push(blk);
|
||||
}
|
||||
let first_mbr_blkno =
|
||||
@@ -1122,7 +1142,7 @@ pub fn decode_wal_record(checkpoint: &mut CheckPoint, record: Bytes) -> DecodedW
|
||||
let mut blk = DecodedBkpBlock::new();
|
||||
blk.forknum = pg_constants::PG_MXACT_MEMBERS_FORKNUM;
|
||||
blk.blkno = blkno;
|
||||
blk.will_init = true;
|
||||
blk.will_drop = true;
|
||||
blocks.push(blk);
|
||||
}
|
||||
} else {
|
||||
|
||||
@@ -201,7 +201,7 @@ fn walreceiver_main(
|
||||
|
||||
let new_checkpoint_bytes = encode_checkpoint(checkpoint);
|
||||
if new_checkpoint_bytes != old_checkpoint_bytes {
|
||||
timeline.put_page_image(checkpoint_tag, Lsn(0), new_checkpoint_bytes);
|
||||
timeline.put_page_image(checkpoint_tag, Lsn(0), new_checkpoint_bytes)?;
|
||||
}
|
||||
// Now that this record has been handled, let the page cache know that
|
||||
// it is up-to-date to this LSN
|
||||
|
||||
@@ -445,9 +445,6 @@ impl PostgresRedoManagerInternal {
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if info == pg_constants::XLOG_MULTIXACT_TRUNCATE_ID {
|
||||
// empty page image indicates that this SLRU page is truncated and can be removed by GC
|
||||
page.clear();
|
||||
} else {
|
||||
panic!();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user