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

More work on compaction, and resurrect some unit tests

Browse Source
This commit is contained in:
Heikki Linnakangas
2022-03-08 13:27:42 +02:00
parent 28045890eb
commit 798ff26fb0
5 changed files with 179 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pageserver/src/keyspace.rs
View File

@@ -12,6 +12,7 @@ pub const TARGET_FILE_SIZE: usize = (TARGET_FILE_SIZE_BYTES / 8192) as usize;
///
/// Represents a set of Keys, in a compact form.
///
#[derive(Debug, Clone)]
pub struct KeyPartitioning {
accum: Option<Range<Key>>,

25
pageserver/src/layered_repository.rs
View File

@@ -1433,7 +1433,7 @@ impl LayeredTimeline {
Lsn(0)
};
let num_deltas = layers.get_deltas(&img_range, &(img_lsn..lsn))?.len();
let num_deltas = layers.count_deltas(&img_range, &(img_lsn..lsn))?;
info!(
"range {}-{}, has {} deltas on this timeline",
@@ -1889,6 +1889,10 @@ mod tests {
Ok(())
}
// Target file size in the unit tests. In production, the target
// file size is much larger, maybe 1 GB. But a small size makes it
// much faster to exercise all the logic for creating the files,
// garbage collection, compaction etc.
const TEST_FILE_SIZE: usize = 4 * 1024 * 1024;
#[test]
@@ -1940,6 +1944,10 @@ mod tests {
Ok(())
}
//
// Insert a bunch of key-value pairs with increasing keys, checkpoint,
// repeat 100 times.
//
#[test]
fn test_bulk_insert() -> Result<()> {
let repo = RepoHarness::create("test_bulk_insert")?.load();
@@ -1947,10 +1955,12 @@ mod tests {
let mut lsn = Lsn(0x10);
let mut parts = KeyPartitioning::new();
let mut test_key = Key::from_hex("012222222233333333444444445500000000").unwrap();
let mut blknum = 0;
for _ in 1..100 {
for _ in 1..10000 {
for _ in 0..50 {
for _ in 0..1000 {
test_key.field6 = blknum;
let writer = tline.writer();
writer.put(
@@ -1961,11 +1971,20 @@ mod tests {
writer.advance_last_record_lsn(lsn);
drop(writer);
parts.add_key(test_key);
lsn = Lsn(lsn.0 + 0x10);
blknum += 1;
}
let cutoff = tline.get_last_record_lsn();
parts.repartition(TEST_FILE_SIZE as u64);
tline.hint_partitioning(parts.clone())?;
tline.update_gc_info(Vec::new(), cutoff);
tline.checkpoint(CheckpointConfig::Forced)?;
tline.compact(TEST_FILE_SIZE)?;
tline.gc()?;
}
Ok(())

21
pageserver/src/layered_repository/layer_map.rs
View File

@@ -10,7 +10,7 @@
//! corresponding files are written to disk.
//!
use crate::layered_repository::storage_layer::range_overlaps;
use crate::layered_repository::storage_layer::{range_eq, range_overlaps};
use crate::layered_repository::storage_layer::Layer;
use crate::layered_repository::InMemoryLayer;
use crate::repository::Key;
@@ -331,12 +331,12 @@ impl LayerMap {
Ok(ranges)
}
pub fn get_deltas(
pub fn count_deltas(
&self,
key_range: &Range<Key>,
lsn_range: &Range<Lsn>,
) -> Result<Vec<Arc<dyn Layer>>> {
let mut deltas = Vec::new();
) -> Result<usize> {
let mut result = 0;
for l in self.historic_layers.iter() {
if !l.is_incremental() {
continue;
@@ -347,9 +347,18 @@ impl LayerMap {
if !range_overlaps(&l.get_key_range(), key_range) {
continue;
}
deltas.push(Arc::clone(l));
// We ignore level0 delta layers. Unless the whole keyspace fits
// into one partition
if !range_eq(key_range, &(Key::MIN..Key::MAX)) &&
range_eq(&l.get_key_range(), &(Key::MIN..Key::MAX))
{
continue;
}
result += 1;
}
Ok(deltas)
Ok(result)
}
pub fn get_level0_deltas(&self) -> Result<Vec<Arc<dyn Layer>>> {

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

@@ -24,6 +24,13 @@ where
}
}
pub fn range_eq<T>(a: &Range<T>, b: &Range<T>) -> bool
where
T: PartialEq<T>,
{
a.start == b.start && a.end == b.end
}
/// FIXME
/// Struct used to communicate across calls to 'get_page_reconstruct_data'.
///

231
pageserver/src/repository.rs
View File

@@ -59,6 +59,17 @@ impl Key {
}
key
}
pub fn from_array(b: [u8; 18]) -> Self {
Key {
field1: b[0],
field2: u32::from_be_bytes(b[1..5].try_into().unwrap()),
field3: u32::from_be_bytes(b[5..9].try_into().unwrap()),
field4: u32::from_be_bytes(b[9..13].try_into().unwrap()),
field5: b[13],
field6: u32::from_be_bytes(b[14..18].try_into().unwrap()),
}
}
}
@@ -521,28 +532,31 @@ mod tests {
//use postgres_ffi::{pg_constants, xlog_utils::SIZEOF_CHECKPOINT};
//use std::sync::Arc;
use bytes::BytesMut;
use hex_literal::hex;
use lazy_static::lazy_static;
lazy_static! {
static ref TEST_KEY: Key = Key::from_array(hex!("112222222233333333444444445500000001"));
}
#[test]
fn test_basic() -> Result<()> {
let repo = RepoHarness::create("test_basic")?.load();
let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0))?;
#[allow(non_snake_case)]
let TEST_KEY: Key = Key::from_hex("112222222233333333444444445500000001").unwrap();
let writer = tline.writer();
writer.put(TEST_KEY, Lsn(0x10), Value::Image(TEST_IMG("foo at 0x10")))?;
writer.put(*TEST_KEY, Lsn(0x10), Value::Image(TEST_IMG("foo at 0x10")))?;
writer.advance_last_record_lsn(Lsn(0x10));
drop(writer);
let writer = tline.writer();
writer.put(TEST_KEY, Lsn(0x20), Value::Image(TEST_IMG("foo at 0x20")))?;
writer.put(*TEST_KEY, Lsn(0x20), Value::Image(TEST_IMG("foo at 0x20")))?;
writer.advance_last_record_lsn(Lsn(0x20));
drop(writer);
assert_eq!(tline.get(TEST_KEY, Lsn(0x10))?, TEST_IMG("foo at 0x10"));
assert_eq!(tline.get(TEST_KEY, Lsn(0x1f))?, TEST_IMG("foo at 0x10"));
assert_eq!(tline.get(TEST_KEY, Lsn(0x20))?, TEST_IMG("foo at 0x20"));
assert_eq!(tline.get(*TEST_KEY, Lsn(0x10))?, TEST_IMG("foo at 0x10"));
assert_eq!(tline.get(*TEST_KEY, Lsn(0x1f))?, TEST_IMG("foo at 0x10"));
assert_eq!(tline.get(*TEST_KEY, Lsn(0x20))?, TEST_IMG("foo at 0x20"));
Ok(())
}
@@ -610,123 +624,146 @@ mod tests {
Ok(())
}
/* // FIXME: Garbage collection is broken
#[test]
fn test_prohibit_branch_creation_on_garbage_collected_data() -> Result<()> {
let repo =
RepoHarness::create("test_prohibit_branch_creation_on_garbage_collected_data")?.load_page_repo();
fn make_some_layers<T: Timeline>(tline: &T, start_lsn: Lsn) -> Result<()> {
let mut lsn = start_lsn;
#[allow(non_snake_case)]
{
let writer = tline.writer();
// Create a relation on the timeline
writer.put(*TEST_KEY, lsn, Value::Image(TEST_IMG(&format!("foo at {}", lsn))))?;
writer.advance_last_record_lsn(lsn);
lsn += 0x10;
writer.put(*TEST_KEY, lsn, Value::Image(TEST_IMG(&format!("foo at {}", lsn))))?;
writer.advance_last_record_lsn(lsn);
lsn += 0x10;
}
tline.checkpoint(CheckpointConfig::Forced)?;
{
let writer = tline.writer();
writer.put(*TEST_KEY, lsn, Value::Image(TEST_IMG(&format!("foo at {}", lsn))))?;
writer.advance_last_record_lsn(lsn);
lsn += 0x10;
writer.put(*TEST_KEY, lsn, Value::Image(TEST_IMG(&format!("foo at {}", lsn))))?;
writer.advance_last_record_lsn(lsn);
}
tline.checkpoint(CheckpointConfig::Forced)
}
let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0))?;
make_some_layers(&tline, Lsn(0x20))?;
#[test]
fn test_prohibit_branch_creation_on_garbage_collected_data() -> Result<()> {
let repo =
RepoHarness::create("test_prohibit_branch_creation_on_garbage_collected_data")?.load();
let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0))?;
make_some_layers(tline.as_ref(), Lsn(0x20))?;
// this removes layers before lsn 40 (50 minus 10), so there are two remaining layers, image and delta for 31-50
repo.gc_iteration(Some(TIMELINE_ID), 0x10, false)?;
// this removes layers before lsn 40 (50 minus 10), so there are two remaining layers, image and delta for 31-50
// FIXME: this doesn't actually remove any layer currently, given how the checkpointing
// and compaction works. But it does set the 'cutoff' point so that the cross check
// below should fail.
repo.gc_iteration(Some(TIMELINE_ID), 0x10, false)?;
// try to branch at lsn 25, should fail because we already garbage collected the data
match repo.branch_timeline(TIMELINE_ID, NEW_TIMELINE_ID, Lsn(0x25)) {
Ok(_) => panic!("branching should have failed"),
Err(err) => {
assert!(err.to_string().contains("invalid branch start lsn"));
assert!(err
// try to branch at lsn 25, should fail because we already garbage collected the data
match repo.branch_timeline(TIMELINE_ID, NEW_TIMELINE_ID, Lsn(0x25)) {
Ok(_) => panic!("branching should have failed"),
Err(err) => {
assert!(err.to_string().contains("invalid branch start lsn"));
assert!(err
.source()
.unwrap()
.to_string()
.contains("we might've already garbage collected needed data"))
}
}
Ok(())
}
#[test]
fn test_prohibit_branch_creation_on_pre_initdb_lsn() -> Result<()> {
let repo = RepoHarness::create("test_prohibit_branch_creation_on_pre_initdb_lsn")?.load_page_repo();
Ok(())
}
repo.create_empty_timeline(TIMELINE_ID, Lsn(0x50))?;
// try to branch at lsn 0x25, should fail because initdb lsn is 0x50
match repo.branch_timeline(TIMELINE_ID, NEW_TIMELINE_ID, Lsn(0x25)) {
Ok(_) => panic!("branching should have failed"),
Err(err) => {
assert!(&err.to_string().contains("invalid branch start lsn"));
assert!(&err
#[test]
fn test_prohibit_branch_creation_on_pre_initdb_lsn() -> Result<()> {
let repo = RepoHarness::create("test_prohibit_branch_creation_on_pre_initdb_lsn")?.load();
repo.create_empty_timeline(TIMELINE_ID, Lsn(0x50))?;
// try to branch at lsn 0x25, should fail because initdb lsn is 0x50
match repo.branch_timeline(TIMELINE_ID, NEW_TIMELINE_ID, Lsn(0x25)) {
Ok(_) => panic!("branching should have failed"),
Err(err) => {
assert!(&err.to_string().contains("invalid branch start lsn"));
assert!(&err
.source()
.unwrap()
.to_string()
.contains("is earlier than latest GC horizon"));
}
}
Ok(())
}
#[test]
fn test_prohibit_get_page_at_lsn_for_garbage_collected_pages() -> Result<()> {
let repo =
RepoHarness::create("test_prohibit_get_page_at_lsn_for_garbage_collected_pages")?
.load_page_repo();
Ok(())
}
let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0))?;
make_some_layers(&tline, Lsn(0x20))?;
/*
// FIXME: This currently fails to error out. Calling GC doesn't currently
// remove the old value, we'd need to work a little harder
#[test]
fn test_prohibit_get_for_garbage_collected_data() -> Result<()> {
let repo =
RepoHarness::create("test_prohibit_get_for_garbage_collected_data")?
.load();
repo.gc_iteration(Some(TIMELINE_ID), 0x10, false)?;
let latest_gc_cutoff_lsn = tline.get_latest_gc_cutoff_lsn();
assert!(*latest_gc_cutoff_lsn > Lsn(0x25));
// FIXME: GC is currently disabled, so this still works
/*
match tline.get_page_at_lsn(TESTREL_A, 0, Lsn(0x25)) {
Ok(_) => panic!("request for page should have failed"),
Err(err) => assert!(err.to_string().contains("not found at")),
}
*/
Ok(())
let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0))?;
make_some_layers(tline.as_ref(), Lsn(0x20))?;
repo.gc_iteration(Some(TIMELINE_ID), 0x10, false)?;
let latest_gc_cutoff_lsn = tline.get_latest_gc_cutoff_lsn();
assert!(*latest_gc_cutoff_lsn > Lsn(0x25));
match tline.get(*TEST_KEY, Lsn(0x25)) {
Ok(_) => panic!("request for page should have failed"),
Err(err) => assert!(err.to_string().contains("not found at")),
}
Ok(())
}
*/
#[test]
fn test_retain_data_in_parent_which_is_needed_for_child() -> Result<()> {
let repo =
RepoHarness::create("test_retain_data_in_parent_which_is_needed_for_child")?.load_page_repo();
let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0))?;
make_some_layers(&tline, Lsn(0x20))?;
#[test]
fn test_retain_data_in_parent_which_is_needed_for_child() -> Result<()> {
let repo =
RepoHarness::create("test_retain_data_in_parent_which_is_needed_for_child")?.load();
let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0))?;
make_some_layers(tline.as_ref(), Lsn(0x20))?;
repo.branch_timeline(TIMELINE_ID, NEW_TIMELINE_ID, Lsn(0x40))?;
let newtline = match repo.get_timeline(NEW_TIMELINE_ID)?.local_timeline() {
Some(timeline) => timeline,
None => panic!("Should have a local timeline"),
};
repo.branch_timeline(TIMELINE_ID, NEW_TIMELINE_ID, Lsn(0x40))?;
let newtline = match repo.get_timeline(NEW_TIMELINE_ID)?.local_timeline() {
Some(timeline) => timeline,
None => panic!("Should have a local timeline"),
};
// this removes layers before lsn 40 (50 minus 10), so there are two remaining layers, image and delta for 31-50
repo.gc_iteration(Some(TIMELINE_ID), 0x10, false)?;
assert!(newtline.get_page_at_lsn(TESTREL_A, 0, Lsn(0x25)).is_ok());
// this removes layers before lsn 40 (50 minus 10), so there are two remaining layers, image and delta for 31-50
repo.gc_iteration(Some(TIMELINE_ID), 0x10, false)?;
assert!(newtline.get(*TEST_KEY, Lsn(0x25)).is_ok());
Ok(())
}
Ok(())
}
#[test]
fn test_parent_keeps_data_forever_after_branching() -> Result<()> {
let repo = RepoHarness::create("test_parent_keeps_data_forever_after_branching")?.load();
let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0))?;
make_some_layers(tline.as_ref(), Lsn(0x20))?;
#[test]
fn test_parent_keeps_data_forever_after_branching() -> Result<()> {
let harness = RepoHarness::create("test_parent_keeps_data_forever_after_branching")?;
let repo = harness.load_page_repo();
let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0))?;
make_some_layers(&tline, Lsn(0x20))?;
repo.branch_timeline(TIMELINE_ID, NEW_TIMELINE_ID, Lsn(0x40))?;
let newtline = match repo.get_timeline(NEW_TIMELINE_ID)?.local_timeline() {
Some(timeline) => timeline,
None => panic!("Should have a local timeline"),
};
repo.branch_timeline(TIMELINE_ID, NEW_TIMELINE_ID, Lsn(0x40))?;
let newtline = match repo.get_timeline(NEW_TIMELINE_ID)?.local_timeline() {
Some(timeline) => timeline,
None => panic!("Should have a local timeline"),
};
make_some_layers(newtline.as_ref(), Lsn(0x60))?;
make_some_layers(&newtline, Lsn(0x60))?;
// run gc on parent
repo.gc_iteration(Some(TIMELINE_ID), 0x10, false)?;
// run gc on parent
repo.gc_iteration(Some(TIMELINE_ID), 0x10, false)?;
// Check that the data is still accessible on the branch.
assert_eq!(
newtline.get(*TEST_KEY, Lsn(0x50))?,
TEST_IMG(&format!("foo at {}", Lsn(0x40)))
);
// Check that the data is still accessible on the branch.
assert_eq!(
newtline.get_page_at_lsn(TESTREL_A, 0, Lsn(0x50))?,
TEST_IMG(&format!("foo blk 0 at {}", Lsn(0x40)))
);
Ok(())
}
*/
Ok(())
}
}