From 28045890eb3876fd8f70d771ee8c1d4040b3f843 Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas Date: Tue, 8 Mar 2022 11:06:44 +0200 Subject: [PATCH] Work on compaction. --- pageserver/src/keyspace.rs | 87 ++++ pageserver/src/layered_repository.rs | 377 +++++++----------- .../src/layered_repository/delta_layer.rs | 15 +- .../src/layered_repository/image_layer.rs | 14 +- .../src/layered_repository/inmemory_layer.rs | 12 +- .../src/layered_repository/layer_map.rs | 43 +- .../src/layered_repository/storage_layer.rs | 12 +- pageserver/src/lib.rs | 1 + pageserver/src/pgdatadir_mapping.rs | 112 ++++-- pageserver/src/relish.rs | 132 +----- pageserver/src/repository.rs | 72 +++- 11 files changed, 398 insertions(+), 479 deletions(-) create mode 100644 pageserver/src/keyspace.rs diff --git a/pageserver/src/keyspace.rs b/pageserver/src/keyspace.rs new file mode 100644 index 0000000000..2b490d6ffe --- /dev/null +++ b/pageserver/src/keyspace.rs @@ -0,0 +1,87 @@ +use std::ops::Range; + +use crate::repository::{Key, key_range_size, singleton_range}; + +use postgres_ffi::pg_constants; + +// in # of key-value pairs +// FIXME Size of one segment in pages (128 MB) +pub const TARGET_FILE_SIZE_BYTES: u64 = 128 * 1024 * 1024; +pub const TARGET_FILE_SIZE: usize = (TARGET_FILE_SIZE_BYTES / 8192) as usize; + +/// +/// Represents a set of Keys, in a compact form. +/// +pub struct KeyPartitioning { + accum: Option>, + + ranges: Vec>, + + pub partitions: Vec>>, +} + +impl KeyPartitioning { + + pub fn new() -> Self { + KeyPartitioning { + accum: None, + ranges: Vec::new(), + partitions: Vec::new(), + } + } + + pub fn add_key(&mut self, key: Key) { + self.add_range(singleton_range(key)) + } + + pub fn add_range(&mut self, range: Range) { + match self.accum.as_mut() { + Some(accum) => { + if range.start == accum.end { + accum.end = range.end; + } else { + self.ranges.push(accum.clone()); + *accum = range; + } + }, + None => self.accum = Some(range), + } + } + + pub fn repartition(&mut self, target_size: u64) { + let target_nblocks = (target_size / pg_constants::BLCKSZ as u64) as usize; + if let Some(accum) = self.accum.take() { + self.ranges.push(accum); + } + + self.partitions = Vec::new(); + + let mut current_part = Vec::new(); + let mut current_part_size: usize = 0; + for range in &self.ranges { + let this_size = key_range_size(&range) as usize; + + if current_part_size + this_size > target_nblocks && + !current_part.is_empty() + { + self.partitions.push(current_part); + current_part = Vec::new(); + current_part_size = 0; + } + + let mut remain_size = this_size; + let mut start = range.start; + while remain_size > target_nblocks { + let next = start.add(target_nblocks as u32); + self.partitions.push(vec![start..next]); + start = next; + remain_size -= target_nblocks + } + current_part.push(start..range.end); + current_part_size += remain_size; + } + if !current_part.is_empty() { + self.partitions.push(current_part); + } + } +} diff --git a/pageserver/src/layered_repository.rs b/pageserver/src/layered_repository.rs index e344a79373..5169717818 100644 --- a/pageserver/src/layered_repository.rs +++ b/pageserver/src/layered_repository.rs @@ -22,7 +22,7 @@ use tracing::*; use std::cmp::{min, max, Ordering}; use std::collections::hash_map::Entry; use std::collections::BTreeSet; -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::fs; use std::fs::{File, OpenOptions}; use std::io::Write; @@ -34,6 +34,7 @@ use std::time::Instant; use self::metadata::{metadata_path, TimelineMetadata, METADATA_FILE_NAME}; use crate::config::PageServerConf; +use crate::keyspace::KeyPartitioning; use crate::remote_storage::{schedule_timeline_checkpoint_upload, schedule_timeline_download}; use crate::repository::{ GcResult, Repository, RepositoryTimeline, Timeline, TimelineSyncState, TimelineWriter, @@ -68,7 +69,10 @@ use filename::{DeltaFileName, ImageFileName}; use image_layer::{ImageLayer, ImageLayerWriter}; use inmemory_layer::InMemoryLayer; use layer_map::LayerMap; -use storage_layer::{Layer, ValueReconstructResult, ValueReconstructState, TARGET_FILE_SIZE,TARGET_FILE_SIZE_BYTES}; +use layer_map::SearchResult; +use storage_layer::{Layer, ValueReconstructResult, ValueReconstructState}; + +use crate::keyspace::TARGET_FILE_SIZE_BYTES; // re-export this function so that page_cache.rs can use it. pub use crate::layered_repository::ephemeral_file::writeback as writeback_ephemeral_file; @@ -738,6 +742,8 @@ pub struct LayeredTimeline { // garbage collecting data that is still needed by the child timelines. gc_info: RwLock, + partitioning: RwLock>, + // It may change across major versions so for simplicity // keep it after running initdb for a timeline. // It is needed in checks when we want to error on some operations @@ -794,14 +800,11 @@ impl Timeline for LayeredTimeline { debug_assert!(lsn <= self.get_last_record_lsn()); let mut reconstruct_state = ValueReconstructState { - key, - lsn, records: Vec::new(), img: None, // FIXME: check page cache and put the img here - request_lsn: lsn, }; - self.get_reconstruct_data(&mut reconstruct_state)?; + self.get_reconstruct_data(key, lsn, &mut reconstruct_state)?; self.reconstruct_value(key, lsn, reconstruct_state) } @@ -823,11 +826,6 @@ impl Timeline for LayeredTimeline { } } - // Entry point for forced image creation. Only used by tests at the moment. - fn create_images(&self, threshold: usize) -> Result<()> { - self.create_image_layers(threshold) - } - /// /// Validate lsn against initdb_lsn and latest_gc_cutoff_lsn. /// @@ -861,6 +859,11 @@ impl Timeline for LayeredTimeline { self.disk_consistent_lsn.load() } + fn hint_partitioning(&self, partitioning: KeyPartitioning) -> Result<()> { + self.partitioning.write().unwrap().replace(partitioning); + Ok(()) + } + fn writer<'a>(&'a self) -> Box { Box::new(LayeredTimelineWriter { tl: self, @@ -909,6 +912,7 @@ impl LayeredTimeline { retain_lsns: Vec::new(), cutoff: Lsn(0), }), + partitioning: RwLock::new(None), latest_gc_cutoff_lsn: RwLock::new(metadata.latest_gc_cutoff_lsn()), initdb_lsn: metadata.initdb_lsn(), @@ -1002,7 +1006,7 @@ impl LayeredTimeline { /// /// This function takes the current timeline's locked LayerMap as an argument, /// so callers can avoid potential race conditions. - fn get_reconstruct_data(&self, reconstruct_state: &mut ValueReconstructState) -> Result<()> { + fn get_reconstruct_data(&self, key: Key, request_lsn: Lsn, reconstruct_state: &mut ValueReconstructState) -> Result<()> { // Start from the current timeline. let mut timeline_owned; let mut timeline = self; @@ -1013,6 +1017,7 @@ impl LayeredTimeline { let mut prev_lsn = Lsn(u64::MAX); let mut result = ValueReconstructResult::Continue; + let mut cont_lsn = Lsn(request_lsn.0 + 1); loop { // The function should have updated 'state' @@ -1020,28 +1025,28 @@ impl LayeredTimeline { match result { ValueReconstructResult::Complete => return Ok(()), ValueReconstructResult::Continue => { - if prev_lsn <= reconstruct_state.lsn { + if prev_lsn <= cont_lsn { // Didn't make any progress in last iteration. Error out to avoid // getting stuck in the loop. bail!("could not find layer with more data for key {} at LSN {}, request LSN {}", - reconstruct_state.key, - reconstruct_state.lsn, - reconstruct_state.request_lsn) + key, + Lsn(cont_lsn.0 - 1), + request_lsn) } - prev_lsn = reconstruct_state.lsn; + prev_lsn = cont_lsn; } ValueReconstructResult::Missing => { bail!( "could not find data for key {} at LSN {}, for request at LSN {}", - reconstruct_state.key, - reconstruct_state.lsn, - reconstruct_state.request_lsn + key, + cont_lsn, + request_lsn ) } } // Recurse into ancestor if needed - if reconstruct_state.lsn <= timeline.ancestor_lsn { + if Lsn(cont_lsn.0 - 1) <= timeline.ancestor_lsn { //info!("going into ancestor {}", timeline.ancestor_lsn); let ancestor = timeline.get_ancestor_timeline()?; timeline_owned = ancestor; @@ -1055,33 +1060,34 @@ impl LayeredTimeline { // Check the open and frozen in-memory layers first if let Some(open_layer) = &layers.open_layer { let start_lsn = open_layer.get_lsn_range().start; - if reconstruct_state.lsn >= start_lsn { + if cont_lsn >= start_lsn { //info!("CHECKING for {} at {} on open layer {}", reconstruct_state.key, reconstruct_state.lsn, open_layer.filename().display()); - result = open_layer.get_value_reconstruct_data(open_layer.get_lsn_range().start, reconstruct_state)?; + result = open_layer.get_value_reconstruct_data(key, open_layer.get_lsn_range().start..cont_lsn, reconstruct_state)?; + cont_lsn = open_layer.get_lsn_range().start; continue; } } if let Some(frozen_layer) = &layers.frozen_layer { let start_lsn = frozen_layer.get_lsn_range().start; - if reconstruct_state.lsn >= start_lsn { + if cont_lsn >= start_lsn { //info!("CHECKING for {} at {} on frozen layer {}", reconstruct_state.key, reconstruct_state.lsn, frozen_layer.filename().display()); - result = frozen_layer.get_value_reconstruct_data(frozen_layer.get_lsn_range().start, reconstruct_state)?; + result = frozen_layer.get_value_reconstruct_data(key, frozen_layer.get_lsn_range().start..cont_lsn, reconstruct_state)?; + cont_lsn = frozen_layer.get_lsn_range().start; continue; } } - if let Some(search_result) = layers - .search(reconstruct_state.key, reconstruct_state.lsn)? + if let Some(SearchResult { lsn_floor, layer }) = layers + .search(key, cont_lsn)? { //info!("CHECKING for {} at {} on historic layer {}", reconstruct_state.key, reconstruct_state.lsn, layer.filename().display()); - result = search_result - .layer - .get_value_reconstruct_data(search_result.lsn_floor, reconstruct_state)?; + result = layer.get_value_reconstruct_data(key, lsn_floor..cont_lsn, reconstruct_state)?; + cont_lsn = lsn_floor; } else if self.ancestor_timeline.is_some() { // Nothing on this timeline. Traverse to parent result = ValueReconstructResult::Continue; - reconstruct_state.lsn = self.ancestor_lsn; + cont_lsn = Lsn(self.ancestor_lsn.0 + 1); } else { // Nothing found result = ValueReconstructResult::Missing; @@ -1241,9 +1247,9 @@ impl LayeredTimeline { // currently hard-coded at 3. It means, write out a new image layer, // if there are at least three delta layers on top of it. if false { - self.create_image_layers(3)?; + self.compact(TARGET_FILE_SIZE_BYTES as usize)?; } - self.compact_level0()?; + //self.compact_level0()?; } // TODO: We should also compact existing delta layers here. @@ -1357,7 +1363,7 @@ impl LayeredTimeline { Ok(()) } - fn compact(&self) -> Result<()> { + fn compact(&self, target_file_size: usize) -> Result<()> { // // High level strategy for compaction / image creation: // @@ -1392,10 +1398,80 @@ impl LayeredTimeline { // Below are functions compact_level0() and create_image_layers() // but they are a bit ad hoc and don't quite work like it's explained // above. Rewrite it. - todo!() + + let lsn = self.last_record_lsn.load().last; + + // 1. The partitioning was already done by the code in + // pgdatadir_mapping.rs. We just use it here. + let partitioning = self.partitioning.read().unwrap(); + if let Some(partitioning) = partitioning.as_ref() { + // 2. Create new image layers for partitions that have been modified + // "enough". + for partition in &partitioning.partitions { + if self.time_for_new_image_layer(partition, lsn, 3)? { + self.create_image_layer(partition, lsn)?; + } + } + // 3. Compact + self.compact_level0(target_file_size)?; + } else { + info!("Could not compact because no partitioning specified yet"); + } + Ok(()) } - fn compact_level0(&self) -> Result<()> { + // Is it time to create a new image layer for the given partition? + fn time_for_new_image_layer(&self, partition: &Vec>, lsn: Lsn, threshold: usize) -> Result { + let layers = self.layers.lock().unwrap(); + + for part_range in partition { + let image_coverage = layers.image_coverage(&part_range, lsn)?; + for (img_range, last_img) in image_coverage { + let img_lsn = if let Some(ref last_img) = last_img { + last_img.get_lsn_range().end + } else { + Lsn(0) + }; + + let num_deltas = layers.get_deltas(&img_range, &(img_lsn..lsn))?.len(); + + info!( + "range {}-{}, has {} deltas on this timeline", + img_range.start, img_range.end, num_deltas + ); + if num_deltas >= threshold { + return Ok(true); + } + } + } + + Ok(false) + } + + fn create_image_layer(&self, partition: &Vec>, lsn: Lsn) -> Result<()> { + let img_range = partition.first().unwrap().start..partition.last().unwrap().end; + let mut image_layer_writer = + ImageLayerWriter::new(self.conf, self.timelineid, self.tenantid, &img_range, lsn)?; + + for range in partition { + let mut key = range.start; + while key < range.end { + let img = self.get(key, lsn)?; + image_layer_writer.put_image(key, &img)?; + key = key.next(); + } + } + let image_layer = image_layer_writer.finish()?; + + let mut layers = self.layers.lock().unwrap(); + layers.insert_historic(Arc::new(image_layer)); + drop(layers); + // FIXME: need to fsync? + + Ok(()) + } + + fn compact_level0(&self, target_file_size: usize) -> Result<()> { let mut layers = self.layers.lock().unwrap(); // We compact or "shuffle" the level-0 delta layers when 10 have @@ -1433,12 +1509,13 @@ impl LayeredTimeline { }); // Merge the contents of all the input delta layers into a new set - // of delta layers. Each output layer is TARGET_FILE_SIZE_BYTES in - // size, i.e. we don't try to align the layer boundaries with the - // image layers or relation boundaries. TODO: we probably should, - // to allow garbage collection to happen earlier. + // of delta layers, based on the current partitioning. // - // TODO: we should also opportunistically garbage collect what we can. + // TODO: this actually divides the layers into fixed-size chunks, not + // based on the partitioning. + // + // TODO: we should also opportunistically materialize and + // garbage collect what we can. let mut new_layers = Vec::new(); let mut prev_key: Option = None; let mut writer: Option = None; @@ -1448,8 +1525,7 @@ impl LayeredTimeline { if let Some(prev_key) = prev_key { if key != prev_key && writer.is_some() { let size = writer.as_mut().unwrap().size(); - info!("size is now {}", size); - if size > TARGET_FILE_SIZE_BYTES as u64 { + if size > target_file_size as u64 { new_layers.push(writer.take().unwrap().finish(prev_key.next())?); writer = None; } @@ -1487,193 +1563,6 @@ impl LayeredTimeline { Ok(()) } - /// - /// Create new image layers, to allow garbage collection to remove old files. - /// - fn create_image_layers(&self, threshold: usize) -> Result<()> { - let layers = self.layers.lock().unwrap(); - let lsn = self.last_record_lsn.load().last; - let image_coverage = layers.image_coverage(&(Key::MIN..Key::MAX), lsn)?; - drop(layers); - - debug!( - "create_image_layers called with threshold {} at {}", - threshold, lsn - ); - - // For any range where there has been more than 'threshold' - // deltas on top of the last image, create new image. - // - // TODO: Invent a better heuristic. - // - // - // TODO: add heuristics to greedily include more segments in the - // image layer, if it's otherwise very small. - for (key_range, last_img) in image_coverage { - let img_lsn = if let Some(ref last_img) = last_img { - last_img.get_lsn_range().end - } else { - Lsn(0) - }; - - let layers = self.layers.lock().unwrap(); - let num_deltas = layers.get_deltas(&key_range, &(img_lsn..lsn))?.len(); - drop(layers); - - info!( - "range {}-{} has {} deltas on this timeline", - key_range.start, key_range.end, num_deltas - ); - if num_deltas >= threshold { - self.create_image_layers_for_range(&key_range, last_img, lsn)?; - } - } - - Ok(()) - } - - // Get all distinct Keys present in the given key range. - // - // This is used to figure out which parts of the overall keyspace are in use, to - // divide the keyspace into image layers. - // - // TODO: For a large database, this set could be very large. Use ranges or prefixes - // instead of individual keys. - fn collect_keys( - &self, - key_range: &Range, - img: Option>, - lsn: Lsn, - base_keys: &mut HashSet, - delta_keys: &mut HashSet, - ) -> Result<()> { - info!( - "creating image layer for key range {}-{} at {}", - key_range.start, key_range.end, lsn - ); - - let baseline_lsn = if let Some(img) = img { - // This range is covered by an image layer on this timeline. Iterate over all the keys - img.collect_keys(key_range, base_keys)?; - img.get_lsn_range().end - } else if self.ancestor_timeline.is_some() { - // Need to look at the ancestor for this range. - let ancestor = self.get_ancestor_timeline()?; - - ancestor.collect_keys_recurse(key_range, lsn, base_keys)?; - self.ancestor_lsn - } else { - self.initdb_lsn - }; - - // Ok, we have baseline list of keys from the images now - // Add all keys from all the deltas - let deltas = { - let layers = self.layers.lock().unwrap(); - layers.get_deltas(key_range, &(baseline_lsn..lsn))? - }; - - for delta in deltas { - delta.collect_keys(key_range, delta_keys)?; - } - - Ok(()) - } - - fn collect_keys_recurse( - &self, - key_range: &Range, - lsn: Lsn, - keys: &mut HashSet, - ) -> Result<()> { - let layers = self.layers.lock().unwrap(); - let image_coverage = layers.image_coverage(key_range, lsn)?; - drop(layers); - - for (range, last_img) in image_coverage { - let mut tmp_keys = HashSet::new(); - self.collect_keys(&range, last_img, lsn, keys, &mut tmp_keys)?; - keys.extend(tmp_keys); - } - - Ok(()) - } - - /// Create a new set of image layers for the given key range. - fn create_image_layers_for_range( - &self, - key_range: &Range, - img: Option>, - lsn: Lsn, - ) -> Result<()> { - info!( - "creating image layer for {}-{} at {}", - key_range.start, key_range.end, lsn - ); - - // If this gets called multiple times in a row, it's possible that the - // image layer already exists. - let layers = self.layers.lock().unwrap(); - if layers.image_exists(key_range, lsn) { - info!( - "skipping creation of image layer for {}-{} at {} because it already exists", - key_range.start, key_range.end, lsn - ); - return Ok(()); - } - drop(layers); - - let mut base_keys: HashSet = HashSet::new(); - let mut delta_keys: HashSet = HashSet::new(); - self.collect_keys(key_range, img, lsn, &mut base_keys, &mut delta_keys)?; - - if delta_keys.is_empty() { - // Important special case: even though there was delta layers on top of this - // key range, the delta layers didn't contain any updates within the range. - // In that case, if we wrote a new image, it would have identical contents, - // just stamped at a later LSN. Not much point in that. - return Ok(()); - } - - // Divide the key range into roughly TARGET_FILE_SIZE chunks - let mut all_keys_vec: Vec = - base_keys.iter().chain(delta_keys.iter()).cloned().collect(); - all_keys_vec.sort(); - all_keys_vec.dedup(); - - let mut start_idx = 0; - let mut start_key = key_range.start; - while start_idx < all_keys_vec.len() { - let end_idx = std::cmp::min(start_idx + TARGET_FILE_SIZE as usize, all_keys_vec.len()); - let end_key = if end_idx >= all_keys_vec.len() { - key_range.end - } else { - all_keys_vec[end_idx] - }; - - let img_range = start_key..end_key; - - let mut image_layer_writer = - ImageLayerWriter::new(self.conf, self.timelineid, self.tenantid, &img_range, lsn)?; - - for key in all_keys_vec[start_idx..end_idx].iter() { - let img = self.get(*key, lsn)?; - image_layer_writer.put_image(*key, &img)?; - } - let image_layer = image_layer_writer.finish()?; - - let mut layers = self.layers.lock().unwrap(); - layers.insert_historic(Arc::new(image_layer)); - drop(layers); - // FIXME: need to fsync? - - start_idx = end_idx; - start_key = end_key; - } - - Ok(()) - } - /// /// Garbage collect layer files on a timeline that are no longer needed. /// @@ -2000,6 +1889,8 @@ mod tests { Ok(()) } + const TEST_FILE_SIZE: usize = 4 * 1024 * 1024; + #[test] fn test_images() -> Result<()> { let repo = RepoHarness::create("test_images")?.load(); @@ -2014,7 +1905,7 @@ mod tests { drop(writer); tline.checkpoint(CheckpointConfig::Forced)?; - tline.create_images(1)?; + tline.compact(TEST_FILE_SIZE)?; let writer = tline.writer(); writer.put(TEST_KEY, Lsn(0x20), Value::Image(TEST_IMG("foo at 0x20")))?; @@ -2022,7 +1913,7 @@ mod tests { drop(writer); tline.checkpoint(CheckpointConfig::Forced)?; - tline.create_images(1)?; + tline.compact(TEST_FILE_SIZE)?; let writer = tline.writer(); writer.put(TEST_KEY, Lsn(0x30), Value::Image(TEST_IMG("foo at 0x30")))?; @@ -2030,7 +1921,7 @@ mod tests { drop(writer); tline.checkpoint(CheckpointConfig::Forced)?; - tline.create_images(1)?; + tline.compact(TEST_FILE_SIZE)?; let writer = tline.writer(); writer.put(TEST_KEY, Lsn(0x40), Value::Image(TEST_IMG("foo at 0x40")))?; @@ -2038,7 +1929,7 @@ mod tests { drop(writer); tline.checkpoint(CheckpointConfig::Forced)?; - tline.create_images(1)?; + tline.compact(TEST_FILE_SIZE)?; 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")); @@ -2074,7 +1965,7 @@ mod tests { blknum += 1; } tline.checkpoint(CheckpointConfig::Forced)?; - //tline.create_images(1)?; + tline.compact(TEST_FILE_SIZE)?; } Ok(()) @@ -2085,12 +1976,15 @@ mod tests { let repo = RepoHarness::create("test_random_updates")?.load(); let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0))?; - const NUM_KEYS: usize = 20000; + const NUM_KEYS: usize = 1000; let mut lsn = Lsn(0x10); let mut test_key = Key::from_hex("012222222233333333444444445500000000").unwrap(); let mut blknum = 0; + + let mut parts = KeyPartitioning::new(); + for _ in 0..NUM_KEYS { test_key.field6 = blknum; let writer = tline.writer(); @@ -2102,13 +1996,18 @@ mod tests { writer.advance_last_record_lsn(lsn); drop(writer); + parts.add_key(test_key); + lsn = Lsn(lsn.0 + 0x10); blknum += 1; } - for _ in 0..100 { + parts.repartition(TEST_FILE_SIZE as u64); + tline.hint_partitioning(parts)?; + + for _ in 0..50 { for _ in 0..NUM_KEYS { - blknum = thread_rng().gen_range(0..10000); + blknum = thread_rng().gen_range(0..NUM_KEYS) as u32; test_key.field6 = blknum; let writer = tline.writer(); writer.put( @@ -2125,7 +2024,7 @@ mod tests { let cutoff = tline.get_last_record_lsn(); tline.update_gc_info(Vec::new(), cutoff); tline.checkpoint(CheckpointConfig::Forced)?; - tline.create_images(3)?; + tline.compact(TEST_FILE_SIZE)?; tline.gc()?; } diff --git a/pageserver/src/layered_repository/delta_layer.rs b/pageserver/src/layered_repository/delta_layer.rs index e7d916d1e4..3c5acbc53a 100644 --- a/pageserver/src/layered_repository/delta_layer.rs +++ b/pageserver/src/layered_repository/delta_layer.rs @@ -150,13 +150,15 @@ impl Layer for DeltaLayer { fn get_value_reconstruct_data( &self, - lsn_floor: Lsn, + key: Key, + lsn_range: Range, reconstruct_state: &mut ValueReconstructState, ) -> Result { let mut need_image = true; - assert!(self.key_range.contains(&reconstruct_state.key)); + assert!(self.key_range.contains(&key)); + /* FIXME match &reconstruct_state.img { Some((cached_lsn, _)) if &self.lsn_range.end <= cached_lsn => { reconstruct_state.lsn = *cached_lsn; @@ -164,6 +166,7 @@ impl Layer for DeltaLayer { } _ => {} } + */ { // Open the file and lock the metadata in memory @@ -175,16 +178,17 @@ impl Layer for DeltaLayer { .chapter_reader(VALUES_CHAPTER)?; // Scan the page versions backwards, starting from `lsn`. - if let Some(vec_map) = inner.index.get(&reconstruct_state.key) { - let slice = vec_map.slice_range(lsn_floor..=reconstruct_state.lsn); + if let Some(vec_map) = inner.index.get(&key) { + let slice = vec_map.slice_range(lsn_range); for (entry_lsn, pos) in slice.iter().rev() { + /* FIXME match &reconstruct_state.img { Some((cached_lsn, _)) if entry_lsn <= cached_lsn => { - reconstruct_state.lsn = *cached_lsn; return Ok(ValueReconstructResult::Complete); } _ => {} } + */ let val = Value::des(&utils::read_blob_from_chapter(&values_reader, *pos)?)?; match val { @@ -211,7 +215,6 @@ impl Layer for DeltaLayer { // If an older page image is needed to reconstruct the page, let the // caller know. if need_image { - reconstruct_state.lsn = Lsn(self.lsn_range.start.0 - 1); Ok(ValueReconstructResult::Continue) } else { Ok(ValueReconstructResult::Complete) diff --git a/pageserver/src/layered_repository/image_layer.rs b/pageserver/src/layered_repository/image_layer.rs index c13293105a..a8772632ab 100644 --- a/pageserver/src/layered_repository/image_layer.rs +++ b/pageserver/src/layered_repository/image_layer.rs @@ -130,13 +130,14 @@ impl Layer for ImageLayer { /// Look up given page in the file fn get_value_reconstruct_data( &self, - lsn_floor: Lsn, + key: Key, + lsn_range: Range, reconstruct_state: &mut ValueReconstructState, ) -> Result { - assert!(lsn_floor <= self.lsn); - assert!(self.key_range.contains(&reconstruct_state.key)); - assert!(reconstruct_state.lsn >= self.lsn); + assert!(self.key_range.contains(&key)); + assert!(lsn_range.end >= self.lsn); + /* FIXME match reconstruct_state.img { Some((cached_lsn, _)) if self.lsn <= cached_lsn => { reconstruct_state.lsn = cached_lsn; @@ -144,10 +145,11 @@ impl Layer for ImageLayer { } _ => {} } + */ let inner = self.load()?; - if let Some(offset) = inner.index.get(&reconstruct_state.key) { + if let Some(offset) = inner.index.get(&key) { let chapter = inner .book .as_ref() @@ -164,10 +166,8 @@ impl Layer for ImageLayer { let value = Bytes::from(blob); reconstruct_state.img = Some((self.lsn, value)); - reconstruct_state.lsn = self.lsn; Ok(ValueReconstructResult::Complete) } else { - reconstruct_state.lsn = self.lsn; Ok(ValueReconstructResult::Missing) } } diff --git a/pageserver/src/layered_repository/inmemory_layer.rs b/pageserver/src/layered_repository/inmemory_layer.rs index c3ca2fd091..b6f06d143b 100644 --- a/pageserver/src/layered_repository/inmemory_layer.rs +++ b/pageserver/src/layered_repository/inmemory_layer.rs @@ -121,17 +121,18 @@ impl Layer for InMemoryLayer { /// Look up given value in the layer. fn get_value_reconstruct_data( &self, - lsn_floor: Lsn, + key: Key, + lsn_range: Range, reconstruct_state: &mut ValueReconstructState, ) -> Result { - assert!(lsn_floor <= self.start_lsn); + assert!(lsn_range.start <= self.start_lsn); let mut need_image = true; let inner = self.inner.read().unwrap(); // Scan the page versions backwards, starting from `lsn`. - if let Some(vec_map) = inner.index.get(&reconstruct_state.key) { - let slice = vec_map.slice_range(lsn_floor..=reconstruct_state.lsn); + if let Some(vec_map) = inner.index.get(&key) { + let slice = vec_map.slice_range(lsn_range); for (entry_lsn, pos) in slice.iter().rev() { match &reconstruct_state.img { Some((cached_lsn, _)) if entry_lsn <= cached_lsn => { @@ -144,8 +145,6 @@ impl Layer for InMemoryLayer { match value { Value::Image(img) => { reconstruct_state.img = Some((*entry_lsn, img)); - - reconstruct_state.lsn = *entry_lsn; return Ok(ValueReconstructResult::Complete); } Value::WalRecord(rec) => { @@ -166,7 +165,6 @@ impl Layer for InMemoryLayer { // If an older page image is needed to reconstruct the page, let the // caller know. if need_image { - reconstruct_state.lsn = Lsn(self.start_lsn.0 - 1); Ok(ValueReconstructResult::Continue) } else { Ok(ValueReconstructResult::Complete) diff --git a/pageserver/src/layered_repository/layer_map.rs b/pageserver/src/layered_repository/layer_map.rs index 1c8a4c6860..bd687b54ab 100644 --- a/pageserver/src/layered_repository/layer_map.rs +++ b/pageserver/src/layered_repository/layer_map.rs @@ -66,7 +66,7 @@ pub struct SearchResult { } impl LayerMap { - pub fn search(&self, key: Key, lsn: Lsn) -> Result> { + pub fn search(&self, key: Key, end_lsn: Lsn) -> Result> { // linear search // Find the latest image layer that covers the given key let mut latest_img: Option> = None; @@ -80,15 +80,15 @@ impl LayerMap { } let img_lsn = l.get_lsn_range().start; - if img_lsn > lsn { + if img_lsn >= end_lsn { // too new continue; } - if img_lsn == lsn { + if Lsn(img_lsn.0 + 1) == end_lsn { // found exact match return Ok(Some(SearchResult { layer: Arc::clone(l), - lsn_floor: lsn, + lsn_floor: img_lsn, })); } if img_lsn > latest_img_lsn.unwrap_or(Lsn(0)) { @@ -107,19 +107,19 @@ impl LayerMap { continue; } - if l.get_lsn_range().start > lsn { + if l.get_lsn_range().start >= end_lsn { // too new continue; } - if l.get_lsn_range().end > lsn { + if l.get_lsn_range().end >= end_lsn { // this layer contains the requested point in the key/lsn space. // No need to search any further - info!( + trace!( "found layer {} for request on {} at {}", l.filename().display(), key, - lsn + end_lsn ); latest_delta.replace(Arc::clone(l)); break; @@ -135,40 +135,33 @@ impl LayerMap { } } if let Some(l) = latest_delta { - info!( + trace!( "found (old) layer {} for request on {} at {}", l.filename().display(), key, - lsn + end_lsn ); + let lsn_floor = if let Some(latest_img_lsn) = latest_img_lsn { + Lsn(latest_img_lsn.0 + 1) + } else { + l.get_lsn_range().start + }; Ok(Some(SearchResult { - lsn_floor: latest_img_lsn.unwrap_or(l.get_lsn_range().start), + lsn_floor, layer: l, })) } else if let Some(l) = latest_img { - info!("found img layer and no deltas for request on {} at {}", key, lsn); + trace!("found img layer and no deltas for request on {} at {}", key, end_lsn); Ok(Some(SearchResult { lsn_floor: latest_img_lsn.unwrap(), layer: l, })) } else { - info!("no layer found for request on {} at {}", key, lsn); + trace!("no layer found for request on {} at {}", key, end_lsn); Ok(None) } } - pub fn image_exists(&self, key_range: &Range, lsn: Lsn) -> bool { - for l in self.historic_layers.iter() { - if !l.is_incremental() - && l.get_key_range() == *key_range - && l.get_lsn_range().start == lsn - { - return true; - } - } - false - } - /// /// Insert an on-disk layer /// diff --git a/pageserver/src/layered_repository/storage_layer.rs b/pageserver/src/layered_repository/storage_layer.rs index eb31a9c415..c64a472a36 100644 --- a/pageserver/src/layered_repository/storage_layer.rs +++ b/pageserver/src/layered_repository/storage_layer.rs @@ -13,11 +13,6 @@ use std::path::PathBuf; use zenith_utils::lsn::Lsn; -// in # of key-value pairs -// FIXME Size of one segment in pages (128 MB) -pub const TARGET_FILE_SIZE_BYTES: u64 = 128 * 1024 * 1024; -pub const TARGET_FILE_SIZE: u32 = (TARGET_FILE_SIZE_BYTES / 8192) as u32; - pub fn range_overlaps(a: &Range, b: &Range) -> bool where T: PartialOrd, @@ -49,12 +44,8 @@ where /// #[derive(Debug)] pub struct ValueReconstructState { - pub key: Key, - pub lsn: Lsn, pub records: Vec<(Lsn, ZenithWalRecord)>, pub img: Option<(Lsn, Bytes)>, - - pub request_lsn: Lsn, // original request's LSN, for debugging purposes } /// Return value from Layer::get_page_reconstruct_data @@ -117,7 +108,8 @@ pub trait Layer: Send + Sync { /// collect more data. fn get_value_reconstruct_data( &self, - lsn_floor: Lsn, + key: Key, + lsn_range: Range, reconstruct_data: &mut ValueReconstructState, ) -> Result; diff --git a/pageserver/src/lib.rs b/pageserver/src/lib.rs index af4548ebe9..e617316cda 100644 --- a/pageserver/src/lib.rs +++ b/pageserver/src/lib.rs @@ -3,6 +3,7 @@ pub mod branches; pub mod config; pub mod http; pub mod import_datadir; +pub mod keyspace; pub mod layered_repository; pub mod page_cache; pub mod page_service; diff --git a/pageserver/src/pgdatadir_mapping.rs b/pageserver/src/pgdatadir_mapping.rs index 829533d65f..1c1f37ef82 100644 --- a/pageserver/src/pgdatadir_mapping.rs +++ b/pageserver/src/pgdatadir_mapping.rs @@ -7,6 +7,7 @@ //! Clarify that) //! +use crate::keyspace::{KeyPartitioning, TARGET_FILE_SIZE_BYTES}; use crate::relish::*; use crate::repository::*; use crate::repository::{Repository, Timeline}; @@ -30,11 +31,12 @@ where R: Repository, { pub tline: Arc, + pub last_partitioning: Option, } #[derive(Debug, Serialize, Deserialize)] pub struct DbDirectory { - // (dbnode, spcnode) + // (spcnode, dbnode) dbs: HashSet<(Oid, Oid)>, } @@ -67,7 +69,10 @@ static ZERO_PAGE: Bytes = Bytes::from_static(&[0u8; 8192]); impl DatadirTimeline { pub fn new(tline: Arc) -> Self { - DatadirTimeline { tline } + DatadirTimeline { + tline, + last_partitioning: None, + } } //------------------------------------------------------------------------------ @@ -157,7 +162,7 @@ impl DatadirTimeline { } /// Get a list of all existing relations in given tablespace and database. - pub fn list_rels(&self, spcnode: u32, dbnode: u32, lsn: Lsn) -> Result> { + pub fn list_rels(&self, spcnode: Oid, dbnode: Oid, lsn: Lsn) -> Result> { // fetch directory listing let key = rel_dir_to_key(spcnode, dbnode); let buf = self.tline.get(key, lsn)?; @@ -292,6 +297,65 @@ impl DatadirTimeline { //todo!() Ok(0) } + + fn collect_keyspace(&self, lsn: Lsn) -> Result { + // Iterate through key ranges, greedily packing them into partitions + let mut result = KeyPartitioning::new(); + + // Add dbdir + result.add_key(DBDIR_KEY); + + // Fetch list of database dirs and iterate them + let buf = self.tline.get(DBDIR_KEY, lsn)?; + let dbdir = DbDirectory::des(&buf)?; + + let mut dbs: Vec<(Oid, Oid)> = dbdir.dbs.iter().cloned().collect(); + dbs.sort(); + for (spcnode, dbnode) in dbs { + result.add_key(relmap_file_key(spcnode, dbnode)); + let mut rels: Vec = self.list_rels(spcnode, dbnode, lsn)?.iter().cloned().collect(); + rels.sort(); + for rel in rels { + let relsize_key = rel_size_to_key(rel); + let mut buf = self.tline.get(relsize_key, lsn)?; + let relsize = buf.get_u32_le(); + + result.add_range(rel_block_to_key(rel, 0)..rel_block_to_key(rel, relsize)); + result.add_key(relsize_key); + } + } + + // Iterate SLRUs next + for kind in [SlruKind::Clog, SlruKind:: MultiXactMembers, SlruKind::MultiXactOffsets] { + let slrudir_key = slru_dir_to_key(kind); + let buf = self.tline.get(slrudir_key, lsn)?; + let dir = SlruSegmentDirectory::des(&buf)?; + let mut segments: Vec = dir.segments.iter().cloned().collect(); + segments.sort(); + for segno in segments { + let segsize_key = slru_segment_size_to_key(kind, segno); + let mut buf = self.tline.get(segsize_key, lsn)?; + let segsize = buf.get_u32_le(); + + result.add_range(slru_block_to_key(kind, segno, 0)..slru_block_to_key(kind, segno, segsize)); + result.add_key(segsize_key); + } + } + + // Then pg_twophase + let buf = self.tline.get(TWOPHASEDIR_KEY, lsn)?; + let twophase_dir = TwoPhaseDirectory::des(&buf)?; + let mut xids: Vec = twophase_dir.xids.iter().cloned().collect(); + xids.sort(); + for xid in xids { + result.add_key(twophase_file_key(xid)); + } + + result.add_key(CONTROLFILE_KEY); + result.add_key(CHECKPOINT_KEY); + + Ok(result) + } } pub struct DatadirTimelineWriter<'a, R: Repository> { @@ -628,6 +692,8 @@ impl<'a, R: Repository> DatadirTimelineWriter<'a, R> { pub fn finish(self) -> Result<()> { let writer = self.tline.tline.writer(); + let last_partitioning = self.last_partitioning.unwrap_or(Lsn(0)); + for (key, value) in self.pending_updates { writer.put(key, self.lsn, value)?; } @@ -637,6 +703,12 @@ impl<'a, R: Repository> DatadirTimelineWriter<'a, R> { writer.advance_last_record_lsn(self.lsn); + if self.lsn.0 - last_partitioning.0 > TARGET_FILE_SIZE_BYTES / 8 { + let mut partitioning = self.tline.collect_keyspace(self.lsn)?; + partitioning.repartition(TARGET_FILE_SIZE_BYTES); + self.tline.tline.hint_partitioning(partitioning)?; + } + Ok(()) } @@ -971,40 +1043,6 @@ pub fn key_to_slru_block(key: Key) -> Result<(SlruKind, u32, BlockNumber)> { }) } -pub fn key_to_relish_block(key: Key) -> Result<(RelishTag, BlockNumber)> { - // FIXME: there's got to be a bitfields crate or something out there to do this for us.. - - // This only works for keys for blocks that are handled by WalRedo manager. - // TODO: assert that the other fields are zero - - Ok(match key.field1 { - 0x00 => ( - RelishTag::Relation(RelTag { - spcnode: key.field2, - dbnode: key.field3, - relnode: key.field4, - forknum: key.field5, - }), - key.field6, - ), - - 0x01 => ( - RelishTag::Slru { - slru: match key.field2 { - 0x00 => SlruKind::Clog, - 0x01 => SlruKind::MultiXactMembers, - 0x02 => SlruKind::MultiXactOffsets, - _ => bail!("unrecognized slru kind 0x{:02x}", key.field2), - }, - segno: key.field4, - }, - key.field6, - ), - - _ => bail!("unrecognized value kind 0x{:02x}", key.field1), - }) -} - pub fn dbdir_key_range(spcnode: Oid, dbnode: Oid) -> Range { Key { field1: 0x00, diff --git a/pageserver/src/relish.rs b/pageserver/src/relish.rs index 9228829aef..7304c41a0e 100644 --- a/pageserver/src/relish.rs +++ b/pageserver/src/relish.rs @@ -1,4 +1,6 @@ //! +//! FIXME: relishes are obsolete +//! //! Zenith stores PostgreSQL relations, and some other files, in the //! repository. The relations (i.e. tables and indexes) take up most //! of the space in a typical installation, while the other files are @@ -27,107 +29,7 @@ use serde::{Deserialize, Serialize}; use std::fmt; use postgres_ffi::relfile_utils::forknumber_to_name; -use postgres_ffi::{Oid, TransactionId}; - -/// -/// RelishTag identifies one relish. -/// -#[derive(Debug, Clone, Copy, Hash, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)] -pub enum RelishTag { - // Relations correspond to PostgreSQL relation forks. Each - // PostgreSQL relation fork is considered a separate relish. - Relation(RelTag), - - // SLRUs include pg_clog, pg_multixact/members, and - // pg_multixact/offsets. There are other SLRUs in PostgreSQL, but - // they don't need to be stored permanently (e.g. pg_subtrans), - // or we do not support them in zenith yet (pg_commit_ts). - // - // These are currently never requested directly by the compute - // nodes, although in principle that would be possible. However, - // when a new compute node is created, these are included in the - // tarball that we send to the compute node to initialize the - // PostgreSQL data directory. - // - // Each SLRU segment in PostgreSQL is considered a separate - // relish. For example, pg_clog/0000, pg_clog/0001, and so forth. - // - // SLRU segments are divided into blocks, like relations. - Slru { slru: SlruKind, segno: u32 }, - - // Miscellaneous other files that need to be included in the - // tarball at compute node creation. These are non-blocky, and are - // expected to be small. - - // - // FileNodeMap represents PostgreSQL's 'pg_filenode.map' - // files. They are needed to map catalog table OIDs to filenode - // numbers. Usually the mapping is done by looking up a relation's - // 'relfilenode' field in the 'pg_class' system table, but that - // doesn't work for 'pg_class' itself and a few other such system - // relations. See PostgreSQL relmapper.c for details. - // - // Each database has a map file for its local mapped catalogs, - // and there is a separate map file for shared catalogs. - // - // These files are always 512 bytes long (although we don't check - // or care about that in the page server). - // - FileNodeMap { spcnode: Oid, dbnode: Oid }, - - // - // State files for prepared transactions (e.g pg_twophase/1234) - // - TwoPhase { xid: TransactionId }, - - // The control file, stored in global/pg_control - ControlFile, - - // Special entry that represents PostgreSQL checkpoint. It doesn't - // correspond to to any physical file in PostgreSQL, but we use it - // to track fields needed to restore the checkpoint data in the - // control file, when a compute node is created. - Checkpoint, -} - -impl RelishTag { - pub const fn is_blocky(&self) -> bool { - match self { - // These relishes work with blocks - RelishTag::Relation(_) | RelishTag::Slru { slru: _, segno: _ } => true, - - // and these don't - RelishTag::FileNodeMap { - spcnode: _, - dbnode: _, - } - | RelishTag::TwoPhase { xid: _ } - | RelishTag::ControlFile - | RelishTag::Checkpoint => false, - } - } - - // Physical relishes represent files and use - // RelationSizeEntry to track existing and dropped files. - // They can be both blocky and non-blocky. - pub const fn is_physical(&self) -> bool { - match self { - // These relishes represent physical files - RelishTag::Relation(_) - | RelishTag::Slru { .. } - | RelishTag::FileNodeMap { .. } - | RelishTag::TwoPhase { .. } => true, - - // and these don't - RelishTag::ControlFile | RelishTag::Checkpoint => false, - } - } - - // convenience function to check if this relish is a normal relation. - pub const fn is_relation(&self) -> bool { - matches!(self, RelishTag::Relation(_)) - } -} +use postgres_ffi::Oid; /// /// Relation data file segment id throughout the Postgres cluster. @@ -170,34 +72,6 @@ impl fmt::Display for RelTag { } } -/// Display RelTag in the same format that's used in most PostgreSQL debug messages: -/// -/// //[_fsm|_vm|_init] -/// -impl fmt::Display for RelishTag { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - RelishTag::Relation(rel) => rel.fmt(f), - RelishTag::Slru { slru, segno } => { - // e.g. pg_clog/0001 - write!(f, "{}/{:04X}", slru.to_str(), segno) - } - RelishTag::FileNodeMap { spcnode, dbnode } => { - write!(f, "relmapper file for spc {} db {}", spcnode, dbnode) - } - RelishTag::TwoPhase { xid } => { - write!(f, "pg_twophase/{:08X}", xid) - } - RelishTag::ControlFile => { - write!(f, "control file") - } - RelishTag::Checkpoint => { - write!(f, "checkpoint") - } - } - } -} - /// /// Non-relation transaction status files (clog (a.k.a. pg_xact) and /// pg_multixact) in Postgres are handled by SLRU (Simple LRU) buffer, diff --git a/pageserver/src/repository.rs b/pageserver/src/repository.rs index b1aedd984a..ce7bfd5013 100644 --- a/pageserver/src/repository.rs +++ b/pageserver/src/repository.rs @@ -1,3 +1,4 @@ +use crate::keyspace::KeyPartitioning; use crate::walrecord::ZenithWalRecord; use crate::CheckpointConfig; use anyhow::{bail, Result}; @@ -27,26 +28,30 @@ pub struct Key { impl Key { pub fn next(&self) -> Key { + self.add(1) + } + + pub fn add(&self, x: u32) -> Key { let mut key = self.clone(); - let x = key.field6.overflowing_add(1); - key.field6 = x.0; - if x.1 { - let x = key.field5.overflowing_add(1); - key.field5 = x.0; - if x.1 { - let x = key.field4.overflowing_add(1); - key.field4 = x.0; - if x.1 { - let x = key.field3.overflowing_add(1); - key.field3 = x.0; - if x.1 { - let x = key.field2.overflowing_add(1); - key.field2 = x.0; - if x.1 { - let x = key.field1.overflowing_add(1); - key.field1 = x.0; - assert!(!x.1); + let r = key.field6.overflowing_add(x); + key.field6 = r.0; + if r.1 { + let r = key.field5.overflowing_add(1); + key.field5 = r.0; + if r.1 { + let r = key.field4.overflowing_add(1); + key.field4 = r.0; + if r.1 { + let r = key.field3.overflowing_add(1); + key.field3 = r.0; + if r.1 { + let r = key.field2.overflowing_add(1); + key.field2 = r.0; + if r.1 { + let r = key.field1.overflowing_add(1); + key.field1 = r.0; + assert!(!r.1); } } } @@ -56,6 +61,35 @@ impl Key { } } + + +pub fn key_range_size(key_range: &Range) -> u32 { + let start = key_range.start; + let end = key_range.end; + + if end.field1 != start.field1 || + end.field2 != start.field2 || + end.field3 != start.field3 || + end.field4 != start.field4 + { + return u32::MAX; + } + + let start = (start.field5 as u64) << 32 | start.field6 as u64; + let end = (end.field5 as u64) << 32 | end.field6 as u64; + + let diff = end - start; + if diff > u32::MAX as u64 { + u32::MAX + } else { + diff as u32 + } +} + +pub fn singleton_range(key: Key) -> Range { + key..key.next() +} + impl fmt::Display for Key { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!( @@ -323,7 +357,7 @@ pub trait Timeline: Send + Sync { /// know anything about them here in the repository. fn checkpoint(&self, cconf: CheckpointConfig) -> Result<()>; - fn create_images(&self, threshold: usize) -> Result<()>; + fn hint_partitioning(&self, partitioning: KeyPartitioning) -> Result<()>; /// /// Check that it is valid to request operations with that lsn.