From 798ff26fb0f1c381d022b041b322575ae1e6081b Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas Date: Tue, 8 Mar 2022 13:27:42 +0200 Subject: [PATCH] More work on compaction, and resurrect some unit tests --- pageserver/src/keyspace.rs | 1 + pageserver/src/layered_repository.rs | 25 +- .../src/layered_repository/layer_map.rs | 21 +- .../src/layered_repository/storage_layer.rs | 7 + pageserver/src/repository.rs | 231 ++++++++++-------- 5 files changed, 179 insertions(+), 106 deletions(-) diff --git a/pageserver/src/keyspace.rs b/pageserver/src/keyspace.rs index 2b490d6ffe..4e1ebdd32a 100644 --- a/pageserver/src/keyspace.rs +++ b/pageserver/src/keyspace.rs @@ -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>, diff --git a/pageserver/src/layered_repository.rs b/pageserver/src/layered_repository.rs index 5169717818..cd1b723651 100644 --- a/pageserver/src/layered_repository.rs +++ b/pageserver/src/layered_repository.rs @@ -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(()) diff --git a/pageserver/src/layered_repository/layer_map.rs b/pageserver/src/layered_repository/layer_map.rs index bd687b54ab..1ad6351539 100644 --- a/pageserver/src/layered_repository/layer_map.rs +++ b/pageserver/src/layered_repository/layer_map.rs @@ -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, lsn_range: &Range, - ) -> Result>> { - let mut deltas = Vec::new(); + ) -> Result { + 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>> { diff --git a/pageserver/src/layered_repository/storage_layer.rs b/pageserver/src/layered_repository/storage_layer.rs index c64a472a36..7ddaf7a581 100644 --- a/pageserver/src/layered_repository/storage_layer.rs +++ b/pageserver/src/layered_repository/storage_layer.rs @@ -24,6 +24,13 @@ where } } +pub fn range_eq(a: &Range, b: &Range) -> bool +where + T: PartialEq, +{ + a.start == b.start && a.end == b.end +} + /// FIXME /// Struct used to communicate across calls to 'get_page_reconstruct_data'. /// diff --git a/pageserver/src/repository.rs b/pageserver/src/repository.rs index ce7bfd5013..1abf2c365a 100644 --- a/pageserver/src/repository.rs +++ b/pageserver/src/repository.rs @@ -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(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(()) + } }