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

Implement garbage collection.

Browse Source 
The counters returned by garbage collection, in GcResult, don't make much
sense with the snapshot files implemention, so I added new counters. That
broke the test_gc test, so I made a copy of it as test_snapfiles_gc based
on the new counters.

Handling relation drops is still not implemented.
This commit is contained in:
Heikki Linnakangas
2021-07-13 14:04:52 +03:00
parent df8e3e1695
commit 0b2ed17f86
7 changed files with 364 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
pageserver/src/layered_repository.rs
View File

@@ -22,13 +22,14 @@ use bytes::Bytes;
use log::*;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::collections::{HashSet, BTreeSet};
use std::collections::{BTreeMap, HashMap};
use std::fs;
use std::fs::File;
use std::io::Write;
use std::ops::Bound::Included;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use std::time::{Duration, Instant};
use crate::repository::{GcResult, History, RelTag, Repository, Timeline, WALRecord};
use crate::restore_local_repo::import_timeline_wal;
@@ -201,6 +202,83 @@ impl LayeredRepository {
Ok(TimelineMetadata::des(&data)?)
}
//
// How garbage collection works
// --------
//
// +--bar------------->
// /
// +----+-----foo---------------->
// /
// ----main--+-------------------------->
// \
// +-----baz-------->
//
//
// 1. Grab a mutex to prevent new timelines from being created
// 2. Scan all timelines, and on each timeline, make note of the
// all the points where other timelines have been branched off.
// We will refrain from removing page versions at those LSNs.
// 3. For each timeline, scan all snapshot files on the timeline.
// Remove all files for which a newer file exists and which
// don't cover any branch point LSNs.
//
// TODO:
// - if a relation has been modified on a child branch, then we
// we don't need to keep that in the parent anymore.
//
// - Currently, this is only triggered manually by the 'do_gc' command.
// There is no background thread to do it automatically.
fn gc_iteration(conf: &'static PageServerConf, horizon: u64) -> Result<GcResult> {
let mut totals: GcResult = Default::default();
let now = Instant::now();
// TODO: grab mutex to prevent new timelines from being created here.
// Scan all timelines for the branch points.
let mut all_branchpoints: BTreeSet<(ZTimelineId, Lsn)> = BTreeSet::new();
// Remember timelineid and its last_record_lsn for each timeline
let mut timelines: Vec<(ZTimelineId, Lsn)> = Vec::new();
let timelines_path = conf.workdir.join("timelines");
for direntry in fs::read_dir(timelines_path)? {
let direntry = direntry?;
if let Some(fname) = direntry.file_name().to_str() {
if let Ok(timelineid) = fname.parse::<ZTimelineId>() {
// Read the metadata of this timeline to get its parent timeline.
let metadata = Self::load_metadata(conf, timelineid)?;
timelines.push((timelineid, metadata.last_record_lsn));
if let Some(ancestor_timeline) = metadata.ancestor_timeline {
all_branchpoints.insert((ancestor_timeline, metadata.ancestor_lsn));
}
}
}
}
// Ok, we now know all the branch points. Iterate through them.
for (timelineid, last_lsn) in timelines {
let branchpoints: Vec<Lsn> = all_branchpoints.range(
(Included((timelineid, Lsn(0))),
Included((timelineid, Lsn(u64::MAX)))))
.map(|&x| x.1)
.collect();
if let Some(cutoff) = last_lsn.checked_sub(horizon) {
let result = SnapshotLayer::gc_timeline(conf, timelineid, branchpoints, cutoff)?;
totals += result;
}
}
totals.elapsed = now.elapsed();
Ok(totals)
}
}
/// LayerMap is a BTreeMap keyed by RelTag and the snapshot file's start LSN.
@@ -368,9 +446,21 @@ impl Timeline for LayeredTimeline {
todo!();
}
fn gc_iteration(&self, _horizon: u64) -> Result<GcResult> {
//TODO
Ok(Default::default())
fn gc_iteration(&self, horizon: u64) -> Result<GcResult> {
// In the layered repository, event to GC a single timeline,
// we have to scan all the timelines to determine what child
// timelines there are, so that we know to retain snapshot
// files that are still needed by the children. So we just do
// GC on the whole repository.
//
// FIXME: This makes writing repeatable tests harder, if
// activity on other timelines can affect the counters that
// we return
// But do flush the in-memory layers to disk first.
self.checkpoint()?;
LayeredRepository::gc_iteration(self.conf, horizon)
}
fn put_wal_record(&self, tag: BufferTag, rec: WALRecord) -> Result<()> {

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

@@ -31,20 +31,21 @@
//!
use crate::layered_repository::storage_layer::Layer;
use crate::layered_repository::storage_layer::PageVersion;
use crate::repository::{RelTag, WALRecord};
use crate::repository::{GcResult, RelTag, WALRecord};
use crate::walredo::WalRedoManager;
use crate::PageServerConf;
use crate::ZTimelineId;
use anyhow::{anyhow, bail, Result};
use bytes::Bytes;
use log::*;
use std::collections::{BTreeMap, HashSet};
use std::collections::{BTreeMap, HashSet, BTreeSet};
use std::fs;
use std::fs::File;
use std::io::Write;
use std::ops::Bound::Included;
use std::path::PathBuf;
use std::sync::Mutex;
use std::time::Instant;
use bookfile::{Book, BookWriter};
@@ -498,4 +499,91 @@ impl SnapshotLayer {
Some((reltag, start_lsn, end_lsn))
}
///
/// Garbage collect snapshot files on a timeline that are no longer needed.
///
/// The caller specifies how much history is needed with the two arguments:
///
/// retain_lsns: keep page a version of each page at these LSNs
/// cutoff: also keep everything newer than this LSN
///
/// The 'retain_lsns' lists is currently used to prevent removing files that
/// are needed by child timelines. In the future, the user might be able to
/// name additional points in time to retain. The caller is responsible for
/// collecting that information.
///
/// The 'cutoff' point is used to retain recent versions that might still be
/// needed by read-only nodes. (As of this writing, the caller just passes
/// the latest LSN subtracted by a constant, and doesn't do anything smart
/// to figure out what read-only nodes might actually need.)
///
/// Currently, we don't make any attempt at removing unneeded page versions
/// within a snapshot file. We can only remove the whole file if it's fully
/// obsolete.
///
pub fn gc_timeline(conf: &'static PageServerConf,
timelineid: ZTimelineId,
retain_lsns: Vec<Lsn>,
cutoff: Lsn) -> Result<GcResult> {
let now = Instant::now();
let mut result: GcResult = Default::default();
// Scan all snapshot files in the directory. For each file, if a newer file
// 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 timeline_path = conf.timeline_path(timelineid);
for direntry in fs::read_dir(timeline_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) {
snapfiles.insert((reltag, start_lsn, end_lsn));
}
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 {
// Is it newer than cutoff point?
if *end_lsn >= cutoff {
result.snapshot_files_needed_by_cutoff += 1;
continue 'outer;
}
// Is it needed by a child branch?
for retain_lsn in &retain_lsns {
// FIXME: are the bounds inclusive or exclusive?
if *start_lsn <= *retain_lsn && *retain_lsn <= *end_lsn {
result.snapshot_files_needed_by_branches += 1;
continue 'outer;
}
}
// 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() {
// 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);
info!("garbage collecting {}", path.display());
fs::remove_file(path)?;
result.snapshot_files_removed += 1;
}
result.elapsed = now.elapsed();
Ok(result)
}
}

2
pageserver/src/lib.rs
View File

@@ -192,7 +192,7 @@ impl fmt::Display for ZId {
/// is separate from PostgreSQL timelines, and doesn't have those
/// limitations. A zenith timeline is identified by a 128-bit ID, which
/// is usually printed out as a hex string.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct ZTimelineId(ZId);
impl FromStr for ZTimelineId {

70
pageserver/src/page_service.rs
View File

@@ -543,54 +543,21 @@ impl postgres_backend::Handler for PageServerHandler {
let result = timeline.gc_iteration(gc_horizon, true)?;
pgb.write_message_noflush(&BeMessage::RowDescription(&[
RowDescriptor {
name: b"n_relations",
typoid: 20,
typlen: 8,
..Default::default()
},
RowDescriptor {
name: b"truncated",
typoid: 20,
typlen: 8,
..Default::default()
},
RowDescriptor {
name: b"deleted",
typoid: 20,
typlen: 8,
..Default::default()
},
RowDescriptor {
name: b"prep_deleted",
typoid: 20,
typlen: 8,
..Default::default()
},
RowDescriptor {
name: b"slru_deleted",
typoid: 20,
typlen: 8,
..Default::default()
},
RowDescriptor {
name: b"chkp_deleted",
typoid: 20,
typlen: 8,
..Default::default()
},
RowDescriptor {
name: b"dropped",
typoid: 20,
typlen: 8,
..Default::default()
},
RowDescriptor {
name: b"elapsed",
typoid: 20,
typlen: 8,
..Default::default()
},
RowDescriptor::int8_col(b"n_relations"),
RowDescriptor::int8_col(b"truncated"),
RowDescriptor::int8_col(b"deleted"),
RowDescriptor::int8_col(b"prep_deleted"),
RowDescriptor::int8_col(b"slru_deleted"),
RowDescriptor::int8_col(b"chkp_deleted"),
RowDescriptor::int8_col(b"dropped"),
RowDescriptor::int8_col(b"snapshot_files_total"),
RowDescriptor::int8_col(b"snapshot_files_needed_by_cutoff"),
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"elapsed"),
]))?
.write_message_noflush(&BeMessage::DataRow(&[
Some(&result.n_relations.to_string().as_bytes()),
@@ -600,6 +567,13 @@ impl postgres_backend::Handler for PageServerHandler {
Some(&result.slru_deleted.to_string().as_bytes()),
Some(&result.chkp_deleted.to_string().as_bytes()),
Some(&result.dropped.to_string().as_bytes()),
Some(&result.snapshot_files_total.to_string().as_bytes()),
Some(&result.snapshot_files_needed_by_cutoff.to_string().as_bytes()),
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.elapsed.as_millis().to_string().as_bytes()),
]))?
.write_message(&BeMessage::CommandComplete(b"SELECT 1"))?;

30
pageserver/src/repository.rs
View File

@@ -9,6 +9,8 @@ use postgres_ffi::TransactionId;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::iter::Iterator;
use std::fmt;
use std::ops::AddAssign;
use std::sync::Arc;
use std::time::Duration;
use zenith_utils::lsn::Lsn;
@@ -39,6 +41,8 @@ pub trait Repository: Send + Sync {
///
#[derive(Default)]
pub struct GcResult {
// FIXME: These counters make sense for the ObjectRepository. They are not used
// by the LayeredRepository.
pub n_relations: u64,
pub inspected: u64,
pub truncated: u64,
@@ -47,9 +51,35 @@ pub struct GcResult {
pub slru_deleted: u64, // SLRU (clog, multixact)
pub chkp_deleted: u64, // Checkpoints
pub dropped: u64,
// These are used for the LayeredRepository instead
pub snapshot_files_total: u64,
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 elapsed: Duration,
}
impl AddAssign for GcResult {
fn add_assign(&mut self, other: Self) {
self.n_relations += other.n_relations;
self.truncated += other.truncated;
self.deleted += other.deleted;
self.dropped += other.dropped;
self.snapshot_files_total += other.snapshot_files_total;
self.snapshot_files_needed_by_cutoff += other.snapshot_files_needed_by_cutoff;
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.elapsed += other.elapsed;
}
}
pub trait Timeline: Send + Sync {
//------------------------------------------------------------------------------
// Public GET functions

111
test_runner/batch_others/test_snapfiles_gc.py Normal file
View File

@@ -0,0 +1,111 @@
from contextlib import closing
import psycopg2.extras
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))
#
# Test Garbage Collection of old snapshot files
#
# This test is pretty tightly coupled with the current implementation of layered
# storage, in layered_repository.rs.
#
def test_snapfiles_gc(zenith_cli, pageserver, postgres, pg_bin):
zenith_cli.run(["branch", "test_snapfiles_gc", "empty"])
pg = postgres.create_start('test_snapfiles_gc')
with closing(pg.connect()) as conn:
with conn.cursor() as cur:
with closing(pageserver.connect()) as psconn:
with psconn.cursor(cursor_factory = psycopg2.extras.DictCursor) as pscur:
# Get the timeline ID of our branch. We need it for the 'do_gc' command
cur.execute("SHOW zenith.zenith_timeline")
timeline = cur.fetchone()[0]
# Create a test table
cur.execute("CREATE TABLE foo(x integer)")
# Run GC, to clear out any garbage left behind in the catalogs by
# the CREATE TABLE command. We want to have a clean slate with no garbage
# before running the actual tests below, otherwise the counts won't match
# what we expect.
#
# Also run vacuum first to make it less likely that autovacuum or pruning
# kicks in and confuses our numbers.
cur.execute("VACUUM")
print("Running GC before test")
pscur.execute(f"do_gc {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);
# remember the number of files
snapshot_files_total = row['snapshot_files_total']
assert snapshot_files_total > 0
# Insert a row. The first insert will also create a metadata entry for the
# relation, with size == 1 block. Hence, bump up the expected relation count.
snapshot_files_total += 1;
print("Inserting one row and running GC")
cur.execute("INSERT INTO foo VALUES (1)")
pscur.execute(f"do_gc {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);
assert row['snapshot_files_total'] == snapshot_files_total
assert row['snapshot_files_removed'] == 0
# Insert two more rows and run GC.
# This should create a new snapshot file with the new contents, and
# remove the old one.
print("Inserting two more rows and running GC")
cur.execute("INSERT INTO foo VALUES (2)")
cur.execute("INSERT INTO foo VALUES (3)")
pscur.execute(f"do_gc {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);
assert row['snapshot_files_total'] == snapshot_files_total + 1
assert row['snapshot_files_removed'] == 1
# Do it again. Should again create a new snapshot file and remove old one.
print("Inserting two more rows and running GC")
cur.execute("INSERT INTO foo VALUES (2)")
cur.execute("INSERT INTO foo VALUES (3)")
pscur.execute(f"do_gc {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);
assert row['snapshot_files_total'] == snapshot_files_total + 1
assert row['snapshot_files_removed'] == 1
# Run GC again, with no changes in the database. Should not remove anything.
print("Run GC again, with nothing to do")
pscur.execute(f"do_gc {timeline} 0")
row = pscur.fetchone()
print_gc_result(row);
assert row['snapshot_files_total'] == snapshot_files_total
assert row['snapshot_files_removed'] == 0
#
# Test DROP TABLE checks that relation data and metadata was deleted by GC from object storage
#
print("Drop table and run GC again");
cur.execute("DROP TABLE foo")
pscur.execute(f"do_gc {timeline} 0")
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.
# 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

15
zenith_utils/src/pq_proto.rs
View File

@@ -379,6 +379,21 @@ impl Default for RowDescriptor<'_> {
}
}
impl RowDescriptor<'_> {
/// Convenience function to create a RowDescriptor message for an int8 column
pub const fn int8_col(name: &[u8]) -> RowDescriptor {
RowDescriptor {
name,
tableoid: 0,
attnum: 0,
typoid: 20,
typlen: 8,
typmod: 0,
formatcode: 0,
}
}
}
#[derive(Debug)]
pub struct XLogDataBody<'a> {
pub wal_start: u64,