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

Work on compaction.

Browse Source
This commit is contained in:
Heikki Linnakangas
2022-03-08 11:06:44 +02:00
parent 6127b6638b
commit 28045890eb
11 changed files with 398 additions and 479 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
pageserver/src/keyspace.rs Normal file
View File

@@ -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<Range<Key>>,
ranges: Vec<Range<Key>>,
pub partitions: Vec<Vec<Range<Key>>>,
}
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<Key>) {
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);
}
}
}

377
pageserver/src/layered_repository.rs
View File

@@ -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<GcInfo>,
partitioning: RwLock<Option<KeyPartitioning>>,
// 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<dyn TimelineWriter + 'a> {
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<Range<Key>>, lsn: Lsn, threshold: usize) -> Result<bool> {
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<Range<Key>>, 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<Key> = None;
let mut writer: Option<DeltaLayerWriter> = 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<Key>,
img: Option<Arc<dyn Layer>>,
lsn: Lsn,
base_keys: &mut HashSet<Key>,
delta_keys: &mut HashSet<Key>,
) -> 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<Key>,
lsn: Lsn,
keys: &mut HashSet<Key>,
) -> 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<Key>,
img: Option<Arc<dyn Layer>>,
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<Key> = HashSet::new();
let mut delta_keys: HashSet<Key> = 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<Key> =
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()?;
}

15
pageserver/src/layered_repository/delta_layer.rs
View File

@@ -150,13 +150,15 @@ impl Layer for DeltaLayer {
fn get_value_reconstruct_data(
&self,
lsn_floor: Lsn,
key: Key,
lsn_range: Range<Lsn>,
reconstruct_state: &mut ValueReconstructState,
) -> Result<ValueReconstructResult> {
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)

14
pageserver/src/layered_repository/image_layer.rs
View File

@@ -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<Lsn>,
reconstruct_state: &mut ValueReconstructState,
) -> Result<ValueReconstructResult> {
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)
}
}

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

@@ -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<Lsn>,
reconstruct_state: &mut ValueReconstructState,
) -> Result<ValueReconstructResult> {
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)

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

@@ -66,7 +66,7 @@ pub struct SearchResult {
}
impl LayerMap {
pub fn search(&self, key: Key, lsn: Lsn) -> Result<Option<SearchResult>> {
pub fn search(&self, key: Key, end_lsn: Lsn) -> Result<Option<SearchResult>> {
// linear search
// Find the latest image layer that covers the given key
let mut latest_img: Option<Arc<dyn Layer>> = 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<Key>, 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
///

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

@@ -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<T>(a: &Range<T>, b: &Range<T>) -> bool
where
T: PartialOrd<T>,
@@ -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<Lsn>,
reconstruct_data: &mut ValueReconstructState,
) -> Result<ValueReconstructResult>;

1
pageserver/src/lib.rs
View File

@@ -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;

112
pageserver/src/pgdatadir_mapping.rs
View File

@@ -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<R::Timeline>,
pub last_partitioning: Option<Lsn>,
}
#[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<R: Repository> DatadirTimeline<R> {
pub fn new(tline: Arc<R::Timeline>) -> Self {
DatadirTimeline { tline }
DatadirTimeline {
tline,
last_partitioning: None,
}
}
//------------------------------------------------------------------------------
@@ -157,7 +162,7 @@ impl<R: Repository> DatadirTimeline<R> {
}
/// Get a list of all existing relations in given tablespace and database.
pub fn list_rels(&self, spcnode: u32, dbnode: u32, lsn: Lsn) -> Result<HashSet<RelTag>> {
pub fn list_rels(&self, spcnode: Oid, dbnode: Oid, lsn: Lsn) -> Result<HashSet<RelTag>> {
// fetch directory listing
let key = rel_dir_to_key(spcnode, dbnode);
let buf = self.tline.get(key, lsn)?;
@@ -292,6 +297,65 @@ impl<R: Repository> DatadirTimeline<R> {
//todo!()
Ok(0)
}
fn collect_keyspace(&self, lsn: Lsn) -> Result<KeyPartitioning> {
// 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<RelTag> = 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<u32> = 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<TransactionId> = 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> {
Key {
field1: 0x00,

132
pageserver/src/relish.rs
View File

@@ -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:
///
/// <spcnode>/<dbnode>/<relnode>[_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,

72
pageserver/src/repository.rs
View File

@@ -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<Key>) -> 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..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.