rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-07-08 22:50:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement unlinking relations in layered storage.

Browse Source 
If a relation is dropped, the last snapshot file for it is given the
_DROPPED suffix. The garbage collection knows that it can remove the
file when it's old enough, even if there is no newer file.
This commit is contained in:
Heikki Linnakangas
2021-07-13 15:11:16 +03:00
parent 0b2ed17f86
commit 61761bf1ce
7 changed files with 141 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
pageserver/src/layered_repository.rs
View File

@@ -484,9 +484,11 @@ impl Timeline for LayeredTimeline {
snapfile.put_page_image(tag.blknum, lsn, img)
}
fn put_unlink(&self, _tag: RelTag, _lsn: Lsn) -> Result<()> {
// TODO
Ok(())
fn put_unlink(&self, rel: RelTag, lsn: Lsn) -> Result<()> {
debug!("put_unlink: {} at {}", rel, lsn);
let snapfile = self.get_snapshot_file_for_write(rel, lsn)?;
snapfile.put_unlink(lsn)
}
///
@@ -651,6 +653,12 @@ impl LayeredTimeline {
// FIXME: If there is an entry in memory for an older snapshot file,
// but there is a newere snapshot file on disk, this will incorrectly
// return the older entry from memory.
//
// FIXME: If the relation has been dropped, does this return the right
// thing? The compute node should not normally request dropped relations,
// but if OID wraparound happens the same relfilenode might get reused
// for an unrelated relation.
//
if let Some(layer) = layers.get(tag, lsn) {
trace!("found snapshot file in memory: {}", layer.get_start_lsn());
return Ok(Some((layer.clone(), lsn)));
@@ -757,13 +765,17 @@ impl LayeredTimeline {
// Scan the directory for latest existing file.
// FIXME: if this is truly a new rel, none should exist right?
let start_lsn;
if let Some((_start, end)) = SnapshotLayer::find_latest_snapshot_file(
if let Some((_start, end, dropped)) = SnapshotLayer::find_latest_snapshot_file(
self.conf,
self.timelineid,
tag,
Lsn(u64::MAX),
)? {
start_lsn = end;
if dropped {
start_lsn = lsn;
} else {
start_lsn = end;
}
} else {
start_lsn = lsn;
}

34
pageserver/src/layered_repository/inmemory_layer.rs
View File

@@ -33,6 +33,13 @@ pub struct InMemoryLayer {
///
start_lsn: Lsn,
// FIXME: the three mutex-protected fields below should probably be protected
// by a single mutex.
/// If this relation was dropped, remember when that happened. Lsn(0) means
/// it hasn't been dropped
drop_lsn: Mutex<Lsn>,
///
/// All versions of all pages in the layer are are kept here.
/// Indexed by block number and LSN.
@@ -276,12 +283,26 @@ impl Layer for InMemoryLayer {
Ok(())
}
/// Remember that the relation was truncated at given LSN
fn put_unlink(&self, lsn: Lsn) -> anyhow::Result<()> {
let mut drop_lsn = self.drop_lsn.lock().unwrap();
assert!(*drop_lsn == Lsn(0));
*drop_lsn = lsn;
info!("dropped relation {} at {}", self.tag, lsn);
Ok(())
}
///
/// Write the this in-memory layer to disk, as a snapshot layer.
///
fn freeze(&self, end_lsn: Lsn) -> Result<()> {
let page_versions = self.page_versions.lock().unwrap();
let relsizes = self.relsizes.lock().unwrap();
let drop_lsn = self.drop_lsn.lock().unwrap();
// FIXME: we assume there are no modification in-flight, and that there are no
// changes past 'lsn'.
@@ -289,12 +310,23 @@ impl Layer for InMemoryLayer {
let page_versions = page_versions.clone();
let relsizes = relsizes.clone();
let dropped = *drop_lsn != Lsn(0);
let end_lsn =
if dropped {
assert!(*drop_lsn < end_lsn);
*drop_lsn
} else {
end_lsn
};
let _snapfile = SnapshotLayer::create(
self.conf,
self.timelineid,
self.tag,
self.start_lsn,
end_lsn,
dropped,
page_versions,
relsizes,
)?;
@@ -323,6 +355,7 @@ impl InMemoryLayer {
timelineid,
tag,
start_lsn,
drop_lsn: Mutex::new(Lsn(0)),
page_versions: Mutex::new(BTreeMap::new()),
relsizes: Mutex::new(BTreeMap::new()),
})
@@ -364,6 +397,7 @@ impl InMemoryLayer {
timelineid,
tag: src.get_tag(),
start_lsn: lsn,
drop_lsn: Mutex::new(Lsn(0)),
page_versions: Mutex::new(page_versions),
relsizes: Mutex::new(relsizes),
})

94
pageserver/src/layered_repository/snapshot_layer.rs
View File

@@ -26,6 +26,14 @@
//!
//! 1663_13990_2609_0_000000000169C348_000000000169C349
//!
//! If a relation is dropped, we add a '_DROPPED' to the end of the filename to indicate that.
//! So the above example would become:
//!
//! 1663_13990_2609_0_000000000169C348_000000000169C349_DROPPED
//!
//! The end LSN indicates when it was dropped in that case, we don't store it in the
//! file contents in any way.
//!
//! A snapshot file is constructed using the 'bookfile' crate. Each file consists of two
//! parts: the page versions and the relation sizes. They are stored as separate chapters.
//!
@@ -76,6 +84,8 @@ pub struct SnapshotLayer {
pub start_lsn: Lsn,
pub end_lsn: Lsn,
dropped: bool,
///
/// All versions of all pages in the file are are kept here.
/// Indexed by block number and LSN.
@@ -266,6 +276,10 @@ impl Layer for SnapshotLayer {
bail!("cannot modify historical snapshot file");
}
fn put_unlink(&self, _lsn: Lsn) -> anyhow::Result<()> {
bail!("cannot modify historical snapshot file");
}
fn freeze(&self, _end_lsn: Lsn) -> Result<()> {
bail!("cannot freeze historical snapshot file");
}
@@ -279,6 +293,7 @@ impl SnapshotLayer {
self.tag,
self.start_lsn,
self.end_lsn,
self.dropped,
)
}
@@ -288,15 +303,17 @@ impl SnapshotLayer {
tag: RelTag,
start_lsn: Lsn,
end_lsn: Lsn,
dropped: bool
) -> PathBuf {
let fname = format!(
"{}_{}_{}_{}_{:016X}_{:016X}",
"{}_{}_{}_{}_{:016X}_{:016X}{}",
tag.spcnode,
tag.dbnode,
tag.relnode,
tag.forknum,
u64::from(start_lsn),
u64::from(end_lsn)
u64::from(end_lsn),
if dropped { "_DROPPED" } else { "" },
);
conf.timeline_path(timelineid).join(&fname)
@@ -314,6 +331,7 @@ impl SnapshotLayer {
tag: RelTag,
start_lsn: Lsn,
end_lsn: Lsn,
dropped: bool,
page_versions: BTreeMap<(u32, Lsn), PageVersion>,
relsizes: BTreeMap<Lsn, u32>,
) -> Result<SnapshotLayer> {
@@ -323,6 +341,7 @@ impl SnapshotLayer {
tag: tag,
start_lsn: start_lsn,
end_lsn,
dropped,
page_versions: Mutex::new(page_versions),
relsizes: Mutex::new(relsizes),
};
@@ -371,26 +390,28 @@ impl SnapshotLayer {
timelineid: ZTimelineId,
tag: RelTag,
lsn: Lsn,
) -> Result<Option<(Lsn, Lsn)>> {
) -> Result<Option<(Lsn, Lsn, bool)>> {
// Scan the timeline directory to get all rels in this timeline.
let path = conf.timeline_path(timelineid);
let mut result_start_lsn = Lsn(0);
let mut result_end_lsn = Lsn(0);
let mut result_dropped = false;
for direntry in fs::read_dir(path)? {
let direntry = direntry?;
let fname = direntry.file_name();
let fname = fname.to_str().unwrap();
if let Some((reltag, start_lsn, end_lsn)) = Self::fname_to_tag(fname) {
if let Some((reltag, start_lsn, end_lsn, dropped)) = Self::fname_to_tag(fname) {
if reltag == tag && start_lsn <= lsn && start_lsn > result_start_lsn {
result_start_lsn = start_lsn;
result_end_lsn = end_lsn;
result_dropped = dropped;
}
}
}
if result_start_lsn != Lsn(0) {
Ok(Some((result_start_lsn, result_end_lsn)))
Ok(Some((result_start_lsn, result_end_lsn, result_dropped)))
} else {
Ok(None)
}
@@ -407,10 +428,10 @@ impl SnapshotLayer {
tag: RelTag,
lsn: Lsn,
) -> Result<Option<SnapshotLayer>> {
if let Some((start_lsn, end_lsn)) =
if let Some((start_lsn, end_lsn, dropped)) =
Self::find_latest_snapshot_file(conf, timelineid, tag, lsn)?
{
let snap = Self::load_path(conf, timelineid, tag, start_lsn, end_lsn)?;
let snap = Self::load_path(conf, timelineid, tag, start_lsn, end_lsn, dropped)?;
Ok(Some(snap))
} else {
Ok(None)
@@ -423,8 +444,9 @@ impl SnapshotLayer {
tag: RelTag,
start_lsn: Lsn,
end_lsn: Lsn,
dropped: bool,
) -> Result<SnapshotLayer> {
let path = Self::path_for(conf, timelineid, tag, start_lsn, end_lsn);
let path = Self::path_for(conf, timelineid, tag, start_lsn, end_lsn, dropped);
let file = File::open(&path)?;
let mut book = Book::new(file)?;
@@ -449,6 +471,7 @@ impl SnapshotLayer {
tag,
start_lsn,
end_lsn,
dropped,
page_versions: Mutex::new(page_versions),
relsizes: Mutex::new(relsizes),
})
@@ -470,7 +493,11 @@ impl SnapshotLayer {
let fname = direntry.file_name();
let fname = fname.to_str().unwrap();
if let Some((reltag, _start_lsn, _end_lsn)) = Self::fname_to_tag(fname) {
if let Some((reltag, _start_lsn, _end_lsn, _dropped)) = Self::fname_to_tag(fname) {
// FIXME: skip if it was dropped before the requested LSN. But there is no
// LSN argument
if (spcnode == 0 || reltag.spcnode == spcnode)
&& (dbnode == 0 || reltag.dbnode == dbnode)
{
@@ -481,11 +508,15 @@ impl SnapshotLayer {
Ok(rels)
}
fn fname_to_tag(fname: &str) -> Option<(RelTag, Lsn, Lsn)> {
fn fname_to_tag(fname: &str) -> Option<(RelTag, Lsn, Lsn, bool)> {
// Split the filename into parts
//
// <spcnode>_<dbnode>_<relnode>_<forknum>_<start LSN>_<end LSN>
//
// or if it was dropped:
//
// <spcnode>_<dbnode>_<relnode>_<forknum>_<start LSN>_<end LSN>_DROPPED
//
let mut parts = fname.split('_');
let reltag = RelTag {
@@ -497,7 +528,21 @@ impl SnapshotLayer {
let start_lsn = Lsn::from_hex(parts.next()?).ok()?;
let end_lsn = Lsn::from_hex(parts.next()?).ok()?;
Some((reltag, start_lsn, end_lsn))
let mut dropped = false;
if let Some(suffix) = parts.next() {
if suffix == "DROPPED" {
dropped = true;
} else {
warn!("unrecognized filename in timeline dir: {}", fname);
return None;
}
}
if parts.next().is_some() {
warn!("unrecognized filename in timeline dir: {}", fname);
return None;
}
Some((reltag, start_lsn, end_lsn, dropped))
}
@@ -535,7 +580,7 @@ impl SnapshotLayer {
// exists, we can remove the old one.
// For convenience and speed, slurp the list of files in the directoy into memory first.
let mut snapfiles: BTreeSet<(RelTag, Lsn, Lsn)> = BTreeSet::new();
let mut snapfiles: BTreeSet<(RelTag, Lsn, Lsn, bool)> = BTreeSet::new();
let timeline_path = conf.timeline_path(timelineid);
for direntry in fs::read_dir(timeline_path)? {
@@ -543,17 +588,17 @@ impl SnapshotLayer {
let fname = direntry.file_name();
let fname = fname.to_str().unwrap();
if let Some((reltag, start_lsn, end_lsn)) = Self::fname_to_tag(fname) {
snapfiles.insert((reltag, start_lsn, end_lsn));
if let Some((reltag, start_lsn, end_lsn, dropped)) = Self::fname_to_tag(fname) {
snapfiles.insert((reltag, start_lsn, end_lsn, dropped));
}
result.snapshot_files_total += 1;
}
// Now determine for each file if it needs to be retained
'outer: for (reltag, start_lsn, end_lsn) in &snapfiles {
'outer: for (reltag, start_lsn, end_lsn, dropped) in &snapfiles {
// Is it newer than cutoff point?
if *end_lsn >= cutoff {
if *end_lsn > cutoff {
result.snapshot_files_needed_by_cutoff += 1;
continue 'outer;
}
@@ -567,20 +612,25 @@ impl SnapshotLayer {
}
}
// Is there a later snapshot file for this relation?
if snapfiles.range(
(Included((*reltag, *end_lsn, Lsn(0))),
Included((*reltag, Lsn(u64::MAX), Lsn(0))))).next().is_none() {
// Unless the relation was dropped, is there a later snapshot file for this relation?
if !dropped && snapfiles.range(
(Included((*reltag, *end_lsn, Lsn(0), false)),
Included((*reltag, Lsn(u64::MAX), Lsn(0), true)))).next().is_none() {
// there is no later file, so keep it
result.snapshot_files_not_updated += 1;
continue 'outer;
}
// We didn't find any reason to keep this file, so remove it.
let path = Self::path_for(conf, timelineid, *reltag, *start_lsn, *end_lsn);
let path = Self::path_for(conf, timelineid, *reltag, *start_lsn, *end_lsn, *dropped);
info!("garbage collecting {}", path.display());
fs::remove_file(path)?;
result.snapshot_files_removed += 1;
if *dropped {
result.snapshot_files_dropped += 1;
} else {
result.snapshot_files_removed += 1;
}
}
result.elapsed = now.elapsed();

2
pageserver/src/layered_repository/storage_layer.rs
View File

@@ -46,6 +46,8 @@ pub trait Layer: Send + Sync {
fn put_truncation(&self, lsn: Lsn, relsize: u32) -> anyhow::Result<()>;
fn put_unlink(&self, lsn: Lsn) -> anyhow::Result<()>;
/// Remember new page version, as a WAL record over previous version
fn put_wal_record(&self, blknum: u32, rec: WALRecord) -> Result<()> {
// FIXME: If this is the first version of this page, reconstruct the image

2
pageserver/src/page_service.rs
View File

@@ -556,6 +556,7 @@ impl postgres_backend::Handler for PageServerHandler {
RowDescriptor::int8_col(b"snapshot_files_needed_by_branches"),
RowDescriptor::int8_col(b"snapshot_files_not_updated"),
RowDescriptor::int8_col(b"snapshot_files_removed"),
RowDescriptor::int8_col(b"snapshot_files_dropped"),
RowDescriptor::int8_col(b"elapsed"),
]))?
@@ -573,6 +574,7 @@ impl postgres_backend::Handler for PageServerHandler {
Some(&result.snapshot_files_needed_by_branches.to_string().as_bytes()),
Some(&result.snapshot_files_not_updated.to_string().as_bytes()),
Some(&result.snapshot_files_removed.to_string().as_bytes()),
Some(&result.snapshot_files_dropped.to_string().as_bytes()),
Some(&result.elapsed.as_millis().to_string().as_bytes()),
]))?

4
pageserver/src/repository.rs
View File

@@ -57,7 +57,8 @@ pub struct GcResult {
pub snapshot_files_needed_by_cutoff: u64,
pub snapshot_files_needed_by_branches: u64,
pub snapshot_files_not_updated: u64,
pub snapshot_files_removed: u64,
pub snapshot_files_removed: u64, // # of snapshot files removed because they have been made obsolete by newer snapshot files.
pub snapshot_files_dropped: u64, // # of snapshot files removed because the relation was dropped
pub elapsed: Duration,
}
@@ -75,6 +76,7 @@ impl AddAssign for GcResult {
self.snapshot_files_needed_by_branches += other.snapshot_files_needed_by_branches;
self.snapshot_files_not_updated += other.snapshot_files_not_updated;
self.snapshot_files_removed += other.snapshot_files_removed;
self.snapshot_files_dropped += other.snapshot_files_dropped;
self.elapsed += other.elapsed;
}

18
test_runner/batch_others/test_snapfiles_gc.py
View File

@@ -5,7 +5,7 @@ import time;
pytest_plugins = ("fixtures.zenith_fixtures")
def print_gc_result(row):
print("GC duration {elapsed} ms, total: {snapshot_files_total}, needed_by_cutoff {snapshot_files_needed_by_cutoff}, needed_by_branches: {snapshot_files_needed_by_branches}, not_updated: {snapshot_files_not_updated}, removed: {snapshot_files_removed}".format_map(row))
print("GC duration {elapsed} ms, total: {snapshot_files_total}, needed_by_cutoff {snapshot_files_needed_by_cutoff}, needed_by_branches: {snapshot_files_needed_by_branches}, not_updated: {snapshot_files_not_updated}, removed: {snapshot_files_removed}, dropped: {snapshot_files_dropped}".format_map(row))
#
@@ -57,6 +57,7 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
print_gc_result(row);
assert row['snapshot_files_total'] == snapshot_files_total
assert row['snapshot_files_removed'] == 0
assert row['snapshot_files_dropped'] == 0
# Insert two more rows and run GC.
# This should create a new snapshot file with the new contents, and
@@ -70,6 +71,7 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
print_gc_result(row);
assert row['snapshot_files_total'] == snapshot_files_total + 1
assert row['snapshot_files_removed'] == 1
assert row['snapshot_files_dropped'] == 0
# Do it again. Should again create a new snapshot file and remove old one.
print("Inserting two more rows and running GC")
@@ -81,6 +83,7 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
print_gc_result(row);
assert row['snapshot_files_total'] == snapshot_files_total + 1
assert row['snapshot_files_removed'] == 1
assert row['snapshot_files_dropped'] == 0
# Run GC again, with no changes in the database. Should not remove anything.
print("Run GC again, with nothing to do")
@@ -89,6 +92,7 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
print_gc_result(row);
assert row['snapshot_files_total'] == snapshot_files_total
assert row['snapshot_files_removed'] == 0
assert row['snapshot_files_dropped'] == 0
#
# Test DROP TABLE checks that relation data and metadata was deleted by GC from object storage
@@ -100,12 +104,12 @@ def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
row = pscur.fetchone()
print_gc_result(row);
# Each relation fork is counted separately, hence 3. But the catalog
# updates also create new snapshot files of the catalogs.
# Each relation fork is counted separately, hence 3.
assert row['snapshot_files_dropped'] == 3
# The catalog updates also create new snapshot files of the catalogs, which
# are counted as 'removed'
assert row['snapshot_files_removed'] > 0
# TODO: perhaps we should count catalog and user relations separately,
# to make this kind of testing more robust
# FIXME: Unlinking relations hasn't been implemented yet
#assert row['snapshot_files_removed'] == 3
#assert row['snapshot_files_removed'] == 5