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

add SegmentTag to Layer queries

Browse Source
This commit is contained in:
Patrick Insinger
2022-01-02 20:35:06 -08:00
parent 066e3f1c69
commit f5432ea1ca
6 changed files with 48 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
pageserver/src/layered_repository.rs
View File

@@ -830,7 +830,7 @@ impl Timeline for LayeredTimeline {
let segsize;
if let Some((layer, lsn)) = self.get_layer_for_read(seg, lsn)? {
segsize = layer.get_seg_size(lsn)?;
segsize = layer.get_seg_size(seg, lsn)?;
trace!("get_seg_size: {} at {} -> {}", seg, lsn, segsize);
} else {
if segno == 0 {
@@ -854,7 +854,7 @@ impl Timeline for LayeredTimeline {
let result;
if let Some((layer, lsn)) = self.get_layer_for_read(seg, lsn)? {
result = layer.get_seg_exists(lsn)?;
result = layer.get_seg_exists(seg, lsn)?;
} else {
result = false;
}
@@ -1867,7 +1867,7 @@ impl LayeredTimeline {
let mut curr_lsn = lsn;
loop {
let result = layer_ref
.get_page_reconstruct_data(blknum, curr_lsn, cached_lsn_opt, &mut data)
.get_page_reconstruct_data(seg, blknum, curr_lsn, cached_lsn_opt, &mut data)
.with_context(|| {
format!(
"Failed to get reconstruct data {} {:?} {} {} {:?}",

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

@@ -193,6 +193,7 @@ impl Layer for DeltaLayer {
/// Look up given page in the cache.
fn get_page_reconstruct_data(
&self,
seg: SegmentTag,
blknum: u32,
lsn: Lsn,
cached_img_lsn: Option<Lsn>,
@@ -200,7 +201,7 @@ impl Layer for DeltaLayer {
) -> Result<PageReconstructResult> {
let mut need_image = true;
assert!(self.seg.blknum_in_seg(blknum));
assert!(seg.blknum_in_seg(blknum));
match &cached_img_lsn {
Some(cached_lsn) if &self.end_lsn <= cached_lsn => {
@@ -219,8 +220,8 @@ impl Layer for DeltaLayer {
.chapter_reader(PAGE_VERSIONS_CHAPTER)?;
// Scan the metadata BTreeMap backwards, starting from the given entry.
let minkey = (self.seg, blknum, Lsn(0));
let maxkey = (self.seg, blknum, lsn);
let minkey = (seg, blknum, Lsn(0));
let maxkey = (seg, blknum, lsn);
let iter = inner
.page_version_metas
.slice_range((Included(&minkey), Included(&maxkey)))
@@ -258,8 +259,8 @@ impl Layer for DeltaLayer {
// If we didn't find any records for this, check if the request is beyond EOF
if need_image
&& reconstruct_data.records.is_empty()
&& self.seg.rel.is_blocky()
&& blknum - self.seg.segno * RELISH_SEG_SIZE >= inner.get_seg_size(self.seg, lsn)?
&& seg.rel.is_blocky()
&& blknum - seg.segno * RELISH_SEG_SIZE >= inner.get_seg_size(seg, lsn)?
{
return Ok(PageReconstructResult::Missing(self.start_lsn));
}
@@ -277,7 +278,7 @@ impl Layer for DeltaLayer {
}
/// Get size of the relation at given LSN
fn get_seg_size(&self, lsn: Lsn) -> Result<u32> {
fn get_seg_size(&self, seg: SegmentTag, lsn: Lsn) -> Result<u32> {
assert!(lsn >= self.start_lsn);
ensure!(
self.seg.rel.is_blocky(),
@@ -286,11 +287,13 @@ impl Layer for DeltaLayer {
// Scan the BTreeMap backwards, starting from the given entry.
let inner = self.load()?;
inner.get_seg_size(self.seg, lsn)
inner.get_seg_size(seg, lsn)
}
/// Does this segment exist at given LSN?
fn get_seg_exists(&self, lsn: Lsn) -> Result<bool> {
fn get_seg_exists(&self, seg: SegmentTag, lsn: Lsn) -> Result<bool> {
assert_eq!(self.seg, seg, "range get_seg_exists not supported"); // TODO
// Is the requested LSN after the rel was dropped?
if self.dropped && lsn >= self.end_lsn {
return Ok(false);

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

@@ -150,6 +150,7 @@ impl Layer for ImageLayer {
/// Look up given page in the file
fn get_page_reconstruct_data(
&self,
seg: SegmentTag,
blknum: u32,
lsn: Lsn,
cached_img_lsn: Option<Lsn>,
@@ -166,11 +167,11 @@ impl Layer for ImageLayer {
let base_blknum = blknum % RELISH_SEG_SIZE;
let blob_range = inner.get_seg_blob_range(self.seg)?;
let blob_range = inner.get_seg_blob_range(seg)?;
let chapter = inner.book.as_ref().unwrap().chapter_reader(BLOB_CHAPTER)?;
let buf = if self.seg.rel.is_blocky() {
let buf = if seg.rel.is_blocky() {
// Check if the request is beyond EOF
if base_blknum >= get_num_blocks(&blob_range) {
return Ok(PageReconstructResult::Missing(lsn));
@@ -196,20 +197,26 @@ impl Layer for ImageLayer {
}
/// Get size of the segment
fn get_seg_size(&self, _lsn: Lsn) -> Result<u32> {
fn get_seg_size(&self, seg: SegmentTag, _lsn: Lsn) -> Result<u32> {
if !self.seg.rel.is_blocky() {
bail!("get_seg_size called for non-blocky segment");
}
let inner = self.load()?;
let blob_range = inner.get_seg_blob_range(self.seg)?;
let blob_range = inner.get_seg_blob_range(seg)?;
Ok(get_num_blocks(&blob_range))
}
/// Does this segment exist at given LSN?
fn get_seg_exists(&self, _lsn: Lsn) -> Result<bool> {
Ok(true)
fn get_seg_exists(&self, seg: SegmentTag, _lsn: Lsn) -> Result<bool> {
let inner = self.load()?;
Ok(inner
.meta
.as_slice()
.binary_search_by_key(&&seg, |(seg, _meta)| seg)
.is_ok())
}
fn unload(&self) -> Result<()> {
@@ -355,13 +362,14 @@ impl ImageLayer {
src: &dyn Layer,
lsn: Lsn,
) -> Result<ImageLayer> {
// TODO needs to become an image of all segments in the layer
let seg = src.get_seg_tag();
let timelineid = timeline.timelineid;
let startblk;
let size;
if seg.rel.is_blocky() {
size = src.get_seg_size(lsn)?;
size = src.get_seg_size(seg, lsn)?;
startblk = seg.segno * RELISH_SEG_SIZE;
} else {
size = 1;

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

@@ -154,11 +154,14 @@ impl Layer for InMemoryLayer {
/// Look up given page in the cache.
fn get_page_reconstruct_data(
&self,
seg: SegmentTag,
blknum: u32,
lsn: Lsn,
cached_img_lsn: Option<Lsn>,
reconstruct_data: &mut PageReconstructData,
) -> Result<PageReconstructResult> {
assert_eq!(self.seg, seg); // TODO
let mut need_image = true;
assert!(self.seg.blknum_in_seg(blknum));
@@ -202,7 +205,7 @@ impl Layer for InMemoryLayer {
if need_image
&& reconstruct_data.records.is_empty()
&& self.seg.rel.is_blocky()
&& blknum - self.seg.segno * RELISH_SEG_SIZE >= self.get_seg_size(lsn)?
&& blknum - self.seg.segno * RELISH_SEG_SIZE >= self.get_seg_size(seg, lsn)?
{
return Ok(PageReconstructResult::Missing(self.start_lsn));
}
@@ -224,7 +227,9 @@ impl Layer for InMemoryLayer {
}
/// Get size of the relation at given LSN
fn get_seg_size(&self, lsn: Lsn) -> Result<u32> {
fn get_seg_size(&self, seg: SegmentTag, lsn: Lsn) -> Result<u32> {
assert_eq!(self.seg, seg);
assert!(lsn >= self.start_lsn);
ensure!(
self.seg.rel.is_blocky(),
@@ -236,7 +241,9 @@ impl Layer for InMemoryLayer {
}
/// Does this segment exist at given LSN?
fn get_seg_exists(&self, lsn: Lsn) -> Result<bool> {
fn get_seg_exists(&self, seg: SegmentTag, lsn: Lsn) -> Result<bool> {
assert_eq!(self.seg, seg);
let inner = self.inner.read().unwrap();
// If the segment created after requested LSN,
@@ -521,7 +528,7 @@ impl InMemoryLayer {
// Copy the segment size at the start LSN from the predecessor layer.
let mut segsizes = VecMap::default();
if seg.rel.is_blocky() {
let size = src.get_seg_size(start_lsn)?;
let size = src.get_seg_size(seg, start_lsn)?;
segsizes.append(start_lsn, size).unwrap();
}

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

@@ -169,7 +169,7 @@ impl LayerMap {
if (request_rel.spcnode == 0 || reltag.spcnode == request_rel.spcnode)
&& (request_rel.dbnode == 0 || reltag.dbnode == request_rel.dbnode)
{
if let Some(exists) = segentry.exists_at_lsn(lsn)? {
if let Some(exists) = segentry.exists_at_lsn(*seg, lsn)? {
rels.insert(seg.rel, exists);
}
}
@@ -177,7 +177,7 @@ impl LayerMap {
}
_ => {
if tag == None {
if let Some(exists) = segentry.exists_at_lsn(lsn)? {
if let Some(exists) = segentry.exists_at_lsn(*seg, lsn)? {
rels.insert(seg.rel, exists);
}
}
@@ -207,7 +207,7 @@ impl LayerMap {
/// to avoid incorrectly making it visible.
pub fn layer_exists_at_lsn(&self, seg: SegmentTag, lsn: Lsn) -> Result<bool> {
Ok(if let Some(segentry) = self.segs.get(&seg) {
segentry.exists_at_lsn(lsn)?.unwrap_or(false)
segentry.exists_at_lsn(seg, lsn)?.unwrap_or(false)
} else {
false
})
@@ -292,9 +292,9 @@ struct SegEntry {
impl SegEntry {
/// Does the segment exist at given LSN?
/// Return None if object is not found in this SegEntry.
fn exists_at_lsn(&self, lsn: Lsn) -> Result<Option<bool>> {
fn exists_at_lsn(&self, seg: SegmentTag, lsn: Lsn) -> Result<Option<bool>> {
if let Some(layer) = self.get(lsn) {
Ok(Some(layer.get_seg_exists(lsn)?))
Ok(Some(layer.get_seg_exists(seg, lsn)?))
} else {
Ok(None)
}

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

@@ -139,6 +139,7 @@ pub trait Layer: Send + Sync {
/// to collect more data.
fn get_page_reconstruct_data(
&self,
seg: SegmentTag,
blknum: u32,
lsn: Lsn,
cached_img_lsn: Option<Lsn>,
@@ -146,10 +147,10 @@ pub trait Layer: Send + Sync {
) -> Result<PageReconstructResult>;
/// Return size of the segment at given LSN. (Only for blocky relations.)
fn get_seg_size(&self, lsn: Lsn) -> Result<u32>;
fn get_seg_size(&self, seg: SegmentTag, lsn: Lsn) -> Result<u32>;
/// Does the segment exist at given LSN? Or was it dropped before it.
fn get_seg_exists(&self, lsn: Lsn) -> Result<bool>;
fn get_seg_exists(&self, seg: SegmentTag, lsn: Lsn) -> Result<bool>;
/// Does this layer only contain some data for the segment (incremental),
/// or does it contain a version of every page? This is important to know