LCOV - code coverage report
Current view: top level - pageserver/src/tenant/storage_layer - delta_layer.rs (source / functions) Coverage Total Hit
Test: 07bee600374ccd486c69370d0972d9035964fe68.info Lines: 86.7 % 1629 1413
Test Date: 2025-02-20 13:11:02 Functions: 71.8 % 163 117

            Line data    Source code
       1              : //! A DeltaLayer represents a collection of WAL records or page images in a range of
       2              : //! LSNs, and in a range of Keys. It is stored on a file on disk.
       3              : //!
       4              : //! Usually a delta layer only contains differences, in the form of WAL records
       5              : //! against a base LSN. However, if a relation extended or a whole new relation
       6              : //! is created, there would be no base for the new pages. The entries for them
       7              : //! must be page images or WAL records with the 'will_init' flag set, so that
       8              : //! they can be replayed without referring to an older page version.
       9              : //!
      10              : //! The delta files are stored in `timelines/<timeline_id>` directory.  Currently,
      11              : //! there are no subdirectories, and each delta file is named like this:
      12              : //!
      13              : //! ```text
      14              : //!    <key start>-<key end>__<start LSN>-<end LSN>
      15              : //! ```
      16              : //!
      17              : //! For example:
      18              : //!
      19              : //! ```text
      20              : //!    000000067F000032BE0000400000000020B6-000000067F000032BE0000400000000030B6__000000578C6B29-0000000057A50051
      21              : //! ```
      22              : //!
      23              : //! Every delta file consists of three parts: "summary", "values", and
      24              : //! "index". The summary is a fixed size header at the beginning of the file,
      25              : //! and it contains basic information about the layer, and offsets to the other
      26              : //! parts. The "index" is a B-tree, mapping from Key and LSN to an offset in the
      27              : //! "values" part.  The actual page images and WAL records are stored in the
      28              : //! "values" part.
      29              : //!
      30              : use crate::config::PageServerConf;
      31              : use crate::context::{PageContentKind, RequestContext, RequestContextBuilder};
      32              : use crate::page_cache::{self, FileId, PAGE_SZ};
      33              : use crate::tenant::blob_io::BlobWriter;
      34              : use crate::tenant::block_io::{BlockBuf, BlockCursor, BlockLease, BlockReader, FileBlockReader};
      35              : use crate::tenant::disk_btree::{
      36              :     DiskBtreeBuilder, DiskBtreeIterator, DiskBtreeReader, VisitDirection,
      37              : };
      38              : use crate::tenant::storage_layer::layer::S3_UPLOAD_LIMIT;
      39              : use crate::tenant::timeline::GetVectoredError;
      40              : use crate::tenant::vectored_blob_io::{
      41              :     BlobFlag, BufView, StreamingVectoredReadPlanner, VectoredBlobReader, VectoredRead,
      42              :     VectoredReadPlanner,
      43              : };
      44              : use crate::virtual_file::owned_buffers_io::io_buf_ext::{FullSlice, IoBufExt};
      45              : use crate::virtual_file::IoBufferMut;
      46              : use crate::virtual_file::{self, MaybeFatalIo, VirtualFile};
      47              : use crate::TEMP_FILE_SUFFIX;
      48              : use crate::{DELTA_FILE_MAGIC, STORAGE_FORMAT_VERSION};
      49              : use anyhow::{bail, ensure, Context, Result};
      50              : use camino::{Utf8Path, Utf8PathBuf};
      51              : use futures::StreamExt;
      52              : use itertools::Itertools;
      53              : use pageserver_api::config::MaxVectoredReadBytes;
      54              : use pageserver_api::key::{Key, DBDIR_KEY, KEY_SIZE};
      55              : use pageserver_api::keyspace::KeySpace;
      56              : use pageserver_api::models::ImageCompressionAlgorithm;
      57              : use pageserver_api::shard::TenantShardId;
      58              : use pageserver_api::value::Value;
      59              : use rand::{distributions::Alphanumeric, Rng};
      60              : use serde::{Deserialize, Serialize};
      61              : use std::collections::{HashMap, VecDeque};
      62              : use std::fs::File;
      63              : use std::io::SeekFrom;
      64              : use std::ops::Range;
      65              : use std::os::unix::fs::FileExt;
      66              : use std::str::FromStr;
      67              : use std::sync::Arc;
      68              : use tokio::sync::OnceCell;
      69              : use tokio_epoll_uring::IoBuf;
      70              : use tracing::*;
      71              : 
      72              : use utils::{
      73              :     bin_ser::BeSer,
      74              :     id::{TenantId, TimelineId},
      75              :     lsn::Lsn,
      76              : };
      77              : 
      78              : use super::{
      79              :     AsLayerDesc, LayerName, OnDiskValue, OnDiskValueIo, PersistentLayerDesc, ResidentLayer,
      80              :     ValuesReconstructState,
      81              : };
      82              : 
      83              : ///
      84              : /// Header stored in the beginning of the file
      85              : ///
      86              : /// After this comes the 'values' part, starting on block 1. After that,
      87              : /// the 'index' starts at the block indicated by 'index_start_blk'
      88              : ///
      89            0 : #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
      90              : pub struct Summary {
      91              :     /// Magic value to identify this as a neon delta file. Always DELTA_FILE_MAGIC.
      92              :     pub magic: u16,
      93              :     pub format_version: u16,
      94              : 
      95              :     pub tenant_id: TenantId,
      96              :     pub timeline_id: TimelineId,
      97              :     pub key_range: Range<Key>,
      98              :     pub lsn_range: Range<Lsn>,
      99              : 
     100              :     /// Block number where the 'index' part of the file begins.
     101              :     pub index_start_blk: u32,
     102              :     /// Block within the 'index', where the B-tree root page is stored
     103              :     pub index_root_blk: u32,
     104              : }
     105              : 
     106              : impl From<&DeltaLayer> for Summary {
     107            0 :     fn from(layer: &DeltaLayer) -> Self {
     108            0 :         Self::expected(
     109            0 :             layer.desc.tenant_shard_id.tenant_id,
     110            0 :             layer.desc.timeline_id,
     111            0 :             layer.desc.key_range.clone(),
     112            0 :             layer.desc.lsn_range.clone(),
     113            0 :         )
     114            0 :     }
     115              : }
     116              : 
     117              : impl Summary {
     118         2188 :     pub(super) fn expected(
     119         2188 :         tenant_id: TenantId,
     120         2188 :         timeline_id: TimelineId,
     121         2188 :         keys: Range<Key>,
     122         2188 :         lsns: Range<Lsn>,
     123         2188 :     ) -> Self {
     124         2188 :         Self {
     125         2188 :             magic: DELTA_FILE_MAGIC,
     126         2188 :             format_version: STORAGE_FORMAT_VERSION,
     127         2188 : 
     128         2188 :             tenant_id,
     129         2188 :             timeline_id,
     130         2188 :             key_range: keys,
     131         2188 :             lsn_range: lsns,
     132         2188 : 
     133         2188 :             index_start_blk: 0,
     134         2188 :             index_root_blk: 0,
     135         2188 :         }
     136         2188 :     }
     137              : }
     138              : 
     139              : // Flag indicating that this version initialize the page
     140              : const WILL_INIT: u64 = 1;
     141              : 
     142              : /// Struct representing reference to BLOB in layers.
     143              : ///
     144              : /// Reference contains BLOB offset, and for WAL records it also contains
     145              : /// `will_init` flag. The flag helps to determine the range of records
     146              : /// that needs to be applied, without reading/deserializing records themselves.
     147            0 : #[derive(Debug, Serialize, Deserialize, Copy, Clone)]
     148              : pub struct BlobRef(pub u64);
     149              : 
     150              : impl BlobRef {
     151      5369047 :     pub fn will_init(&self) -> bool {
     152      5369047 :         (self.0 & WILL_INIT) != 0
     153      5369047 :     }
     154              : 
     155      9582195 :     pub fn pos(&self) -> u64 {
     156      9582195 :         self.0 >> 1
     157      9582195 :     }
     158              : 
     159     12942956 :     pub fn new(pos: u64, will_init: bool) -> BlobRef {
     160     12942956 :         let mut blob_ref = pos << 1;
     161     12942956 :         if will_init {
     162     12939836 :             blob_ref |= WILL_INIT;
     163     12939836 :         }
     164     12942956 :         BlobRef(blob_ref)
     165     12942956 :     }
     166              : }
     167              : 
     168              : pub const DELTA_KEY_SIZE: usize = KEY_SIZE + 8;
     169              : struct DeltaKey([u8; DELTA_KEY_SIZE]);
     170              : 
     171              : /// This is the key of the B-tree index stored in the delta layer. It consists
     172              : /// of the serialized representation of a Key and LSN.
     173              : impl DeltaKey {
     174      4128396 :     fn from_slice(buf: &[u8]) -> Self {
     175      4128396 :         let mut bytes: [u8; DELTA_KEY_SIZE] = [0u8; DELTA_KEY_SIZE];
     176      4128396 :         bytes.copy_from_slice(buf);
     177      4128396 :         DeltaKey(bytes)
     178      4128396 :     }
     179              : 
     180     13377996 :     fn from_key_lsn(key: &Key, lsn: Lsn) -> Self {
     181     13377996 :         let mut bytes: [u8; DELTA_KEY_SIZE] = [0u8; DELTA_KEY_SIZE];
     182     13377996 :         key.write_to_byte_slice(&mut bytes[0..KEY_SIZE]);
     183     13377996 :         bytes[KEY_SIZE..].copy_from_slice(&u64::to_be_bytes(lsn.0));
     184     13377996 :         DeltaKey(bytes)
     185     13377996 :     }
     186              : 
     187      4128396 :     fn key(&self) -> Key {
     188      4128396 :         Key::from_slice(&self.0)
     189      4128396 :     }
     190              : 
     191      4128396 :     fn lsn(&self) -> Lsn {
     192      4128396 :         Lsn(u64::from_be_bytes(self.0[KEY_SIZE..].try_into().unwrap()))
     193      4128396 :     }
     194              : 
     195      5453679 :     fn extract_lsn_from_buf(buf: &[u8]) -> Lsn {
     196      5453679 :         let mut lsn_buf = [0u8; 8];
     197      5453679 :         lsn_buf.copy_from_slice(&buf[KEY_SIZE..]);
     198      5453679 :         Lsn(u64::from_be_bytes(lsn_buf))
     199      5453679 :     }
     200              : }
     201              : 
     202              : /// This is used only from `pagectl`. Within pageserver, all layers are
     203              : /// [`crate::tenant::storage_layer::Layer`], which can hold a [`DeltaLayerInner`].
     204              : pub struct DeltaLayer {
     205              :     path: Utf8PathBuf,
     206              :     pub desc: PersistentLayerDesc,
     207              :     inner: OnceCell<Arc<DeltaLayerInner>>,
     208              : }
     209              : 
     210              : impl std::fmt::Debug for DeltaLayer {
     211            0 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
     212              :         use super::RangeDisplayDebug;
     213              : 
     214            0 :         f.debug_struct("DeltaLayer")
     215            0 :             .field("key_range", &RangeDisplayDebug(&self.desc.key_range))
     216            0 :             .field("lsn_range", &self.desc.lsn_range)
     217            0 :             .field("file_size", &self.desc.file_size)
     218            0 :             .field("inner", &self.inner)
     219            0 :             .finish()
     220            0 :     }
     221              : }
     222              : 
     223              : /// `DeltaLayerInner` is the in-memory data structure associated with an on-disk delta
     224              : /// file.
     225              : pub struct DeltaLayerInner {
     226              :     // values copied from summary
     227              :     index_start_blk: u32,
     228              :     index_root_blk: u32,
     229              : 
     230              :     file: Arc<VirtualFile>,
     231              :     file_id: FileId,
     232              : 
     233              :     layer_key_range: Range<Key>,
     234              :     layer_lsn_range: Range<Lsn>,
     235              : 
     236              :     max_vectored_read_bytes: Option<MaxVectoredReadBytes>,
     237              : }
     238              : 
     239              : impl DeltaLayerInner {
     240            0 :     pub(crate) fn layer_dbg_info(&self) -> String {
     241            0 :         format!(
     242            0 :             "delta {}..{} {}..{}",
     243            0 :             self.key_range().start,
     244            0 :             self.key_range().end,
     245            0 :             self.lsn_range().start,
     246            0 :             self.lsn_range().end
     247            0 :         )
     248            0 :     }
     249              : }
     250              : 
     251              : impl std::fmt::Debug for DeltaLayerInner {
     252            0 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
     253            0 :         f.debug_struct("DeltaLayerInner")
     254            0 :             .field("index_start_blk", &self.index_start_blk)
     255            0 :             .field("index_root_blk", &self.index_root_blk)
     256            0 :             .finish()
     257            0 :     }
     258              : }
     259              : 
     260              : /// Boilerplate to implement the Layer trait, always use layer_desc for persistent layers.
     261              : impl std::fmt::Display for DeltaLayer {
     262            0 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
     263            0 :         write!(f, "{}", self.layer_desc().short_id())
     264            0 :     }
     265              : }
     266              : 
     267              : impl AsLayerDesc for DeltaLayer {
     268            0 :     fn layer_desc(&self) -> &PersistentLayerDesc {
     269            0 :         &self.desc
     270            0 :     }
     271              : }
     272              : 
     273              : impl DeltaLayer {
     274            0 :     pub async fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
     275            0 :         self.desc.dump();
     276            0 : 
     277            0 :         if !verbose {
     278            0 :             return Ok(());
     279            0 :         }
     280              : 
     281            0 :         let inner = self.load(ctx).await?;
     282              : 
     283            0 :         inner.dump(ctx).await
     284            0 :     }
     285              : 
     286         2936 :     fn temp_path_for(
     287         2936 :         conf: &PageServerConf,
     288         2936 :         tenant_shard_id: &TenantShardId,
     289         2936 :         timeline_id: &TimelineId,
     290         2936 :         key_start: Key,
     291         2936 :         lsn_range: &Range<Lsn>,
     292         2936 :     ) -> Utf8PathBuf {
     293         2936 :         let rand_string: String = rand::thread_rng()
     294         2936 :             .sample_iter(&Alphanumeric)
     295         2936 :             .take(8)
     296         2936 :             .map(char::from)
     297         2936 :             .collect();
     298         2936 : 
     299         2936 :         conf.timeline_path(tenant_shard_id, timeline_id)
     300         2936 :             .join(format!(
     301         2936 :                 "{}-XXX__{:016X}-{:016X}.{}.{}",
     302         2936 :                 key_start,
     303         2936 :                 u64::from(lsn_range.start),
     304         2936 :                 u64::from(lsn_range.end),
     305         2936 :                 rand_string,
     306         2936 :                 TEMP_FILE_SUFFIX,
     307         2936 :             ))
     308         2936 :     }
     309              : 
     310              :     ///
     311              :     /// Open the underlying file and read the metadata into memory, if it's
     312              :     /// not loaded already.
     313              :     ///
     314            0 :     async fn load(&self, ctx: &RequestContext) -> Result<&Arc<DeltaLayerInner>> {
     315            0 :         // Quick exit if already loaded
     316            0 :         self.inner
     317            0 :             .get_or_try_init(|| self.load_inner(ctx))
     318            0 :             .await
     319            0 :             .with_context(|| format!("Failed to load delta layer {}", self.path()))
     320            0 :     }
     321              : 
     322            0 :     async fn load_inner(&self, ctx: &RequestContext) -> anyhow::Result<Arc<DeltaLayerInner>> {
     323            0 :         let path = self.path();
     324              : 
     325            0 :         let loaded = DeltaLayerInner::load(&path, None, None, ctx).await?;
     326              : 
     327              :         // not production code
     328            0 :         let actual_layer_name = LayerName::from_str(path.file_name().unwrap()).unwrap();
     329            0 :         let expected_layer_name = self.layer_desc().layer_name();
     330            0 : 
     331            0 :         if actual_layer_name != expected_layer_name {
     332            0 :             println!("warning: filename does not match what is expected from in-file summary");
     333            0 :             println!("actual: {:?}", actual_layer_name.to_string());
     334            0 :             println!("expected: {:?}", expected_layer_name.to_string());
     335            0 :         }
     336              : 
     337            0 :         Ok(Arc::new(loaded))
     338            0 :     }
     339              : 
     340              :     /// Create a DeltaLayer struct representing an existing file on disk.
     341              :     ///
     342              :     /// This variant is only used for debugging purposes, by the 'pagectl' binary.
     343            0 :     pub fn new_for_path(path: &Utf8Path, file: File) -> Result<Self> {
     344            0 :         let mut summary_buf = vec![0; PAGE_SZ];
     345            0 :         file.read_exact_at(&mut summary_buf, 0)?;
     346            0 :         let summary = Summary::des_prefix(&summary_buf)?;
     347              : 
     348            0 :         let metadata = file
     349            0 :             .metadata()
     350            0 :             .context("get file metadata to determine size")?;
     351              : 
     352              :         // This function is never used for constructing layers in a running pageserver,
     353              :         // so it does not need an accurate TenantShardId.
     354            0 :         let tenant_shard_id = TenantShardId::unsharded(summary.tenant_id);
     355            0 : 
     356            0 :         Ok(DeltaLayer {
     357            0 :             path: path.to_path_buf(),
     358            0 :             desc: PersistentLayerDesc::new_delta(
     359            0 :                 tenant_shard_id,
     360            0 :                 summary.timeline_id,
     361            0 :                 summary.key_range,
     362            0 :                 summary.lsn_range,
     363            0 :                 metadata.len(),
     364            0 :             ),
     365            0 :             inner: OnceCell::new(),
     366            0 :         })
     367            0 :     }
     368              : 
     369              :     /// Path to the layer file in pageserver workdir.
     370            0 :     fn path(&self) -> Utf8PathBuf {
     371            0 :         self.path.clone()
     372            0 :     }
     373              : }
     374              : 
     375              : /// A builder object for constructing a new delta layer.
     376              : ///
     377              : /// Usage:
     378              : ///
     379              : /// 1. Create the DeltaLayerWriter by calling DeltaLayerWriter::new(...)
     380              : ///
     381              : /// 2. Write the contents by calling `put_value` for every page
     382              : ///    version to store in the layer.
     383              : ///
     384              : /// 3. Call `finish`.
     385              : ///
     386              : struct DeltaLayerWriterInner {
     387              :     pub path: Utf8PathBuf,
     388              :     timeline_id: TimelineId,
     389              :     tenant_shard_id: TenantShardId,
     390              : 
     391              :     key_start: Key,
     392              :     lsn_range: Range<Lsn>,
     393              : 
     394              :     tree: DiskBtreeBuilder<BlockBuf, DELTA_KEY_SIZE>,
     395              : 
     396              :     blob_writer: BlobWriter<true>,
     397              : 
     398              :     // Number of key-lsns in the layer.
     399              :     num_keys: usize,
     400              : }
     401              : 
     402              : impl DeltaLayerWriterInner {
     403              :     ///
     404              :     /// Start building a new delta layer.
     405              :     ///
     406         2936 :     async fn new(
     407         2936 :         conf: &'static PageServerConf,
     408         2936 :         timeline_id: TimelineId,
     409         2936 :         tenant_shard_id: TenantShardId,
     410         2936 :         key_start: Key,
     411         2936 :         lsn_range: Range<Lsn>,
     412         2936 :         ctx: &RequestContext,
     413         2936 :     ) -> anyhow::Result<Self> {
     414         2936 :         // Create the file initially with a temporary filename. We don't know
     415         2936 :         // the end key yet, so we cannot form the final filename yet. We will
     416         2936 :         // rename it when we're done.
     417         2936 :         //
     418         2936 :         // Note: This overwrites any existing file. There shouldn't be any.
     419         2936 :         // FIXME: throw an error instead?
     420         2936 :         let path =
     421         2936 :             DeltaLayer::temp_path_for(conf, &tenant_shard_id, &timeline_id, key_start, &lsn_range);
     422              : 
     423         2936 :         let mut file = VirtualFile::create(&path, ctx).await?;
     424              :         // make room for the header block
     425         2936 :         file.seek(SeekFrom::Start(PAGE_SZ as u64)).await?;
     426         2936 :         let blob_writer = BlobWriter::new(file, PAGE_SZ as u64);
     427         2936 : 
     428         2936 :         // Initialize the b-tree index builder
     429         2936 :         let block_buf = BlockBuf::new();
     430         2936 :         let tree_builder = DiskBtreeBuilder::new(block_buf);
     431         2936 : 
     432         2936 :         Ok(Self {
     433         2936 :             path,
     434         2936 :             timeline_id,
     435         2936 :             tenant_shard_id,
     436         2936 :             key_start,
     437         2936 :             lsn_range,
     438         2936 :             tree: tree_builder,
     439         2936 :             blob_writer,
     440         2936 :             num_keys: 0,
     441         2936 :         })
     442         2936 :     }
     443              : 
     444              :     ///
     445              :     /// Append a key-value pair to the file.
     446              :     ///
     447              :     /// The values must be appended in key, lsn order.
     448              :     ///
     449      4169428 :     async fn put_value(
     450      4169428 :         &mut self,
     451      4169428 :         key: Key,
     452      4169428 :         lsn: Lsn,
     453      4169428 :         val: Value,
     454      4169428 :         ctx: &RequestContext,
     455      4169428 :     ) -> anyhow::Result<()> {
     456      4169428 :         let (_, res) = self
     457      4169428 :             .put_value_bytes(
     458      4169428 :                 key,
     459      4169428 :                 lsn,
     460      4169428 :                 Value::ser(&val)?.slice_len(),
     461      4169428 :                 val.will_init(),
     462      4169428 :                 ctx,
     463      4169428 :             )
     464      4169428 :             .await;
     465      4169428 :         res
     466      4169428 :     }
     467              : 
     468     12942748 :     async fn put_value_bytes<Buf>(
     469     12942748 :         &mut self,
     470     12942748 :         key: Key,
     471     12942748 :         lsn: Lsn,
     472     12942748 :         val: FullSlice<Buf>,
     473     12942748 :         will_init: bool,
     474     12942748 :         ctx: &RequestContext,
     475     12942748 :     ) -> (FullSlice<Buf>, anyhow::Result<()>)
     476     12942748 :     where
     477     12942748 :         Buf: IoBuf + Send,
     478     12942748 :     {
     479     12942748 :         assert!(
     480     12942748 :             self.lsn_range.start <= lsn,
     481            0 :             "lsn_start={}, lsn={}",
     482              :             self.lsn_range.start,
     483              :             lsn
     484              :         );
     485              :         // We don't want to use compression in delta layer creation
     486     12942748 :         let compression = ImageCompressionAlgorithm::Disabled;
     487     12942748 :         let (val, res) = self
     488     12942748 :             .blob_writer
     489     12942748 :             .write_blob_maybe_compressed(val, ctx, compression)
     490     12942748 :             .await;
     491     12942748 :         let off = match res {
     492     12942748 :             Ok((off, _)) => off,
     493            0 :             Err(e) => return (val, Err(anyhow::anyhow!(e))),
     494              :         };
     495              : 
     496     12942748 :         let blob_ref = BlobRef::new(off, will_init);
     497     12942748 : 
     498     12942748 :         let delta_key = DeltaKey::from_key_lsn(&key, lsn);
     499     12942748 :         let res = self.tree.append(&delta_key.0, blob_ref.0);
     500     12942748 : 
     501     12942748 :         self.num_keys += 1;
     502     12942748 : 
     503     12942748 :         (val, res.map_err(|e| anyhow::anyhow!(e)))
     504     12942748 :     }
     505              : 
     506      4047944 :     fn size(&self) -> u64 {
     507      4047944 :         self.blob_writer.size() + self.tree.borrow_writer().size()
     508      4047944 :     }
     509              : 
     510              :     ///
     511              :     /// Finish writing the delta layer.
     512              :     ///
     513         2884 :     async fn finish(
     514         2884 :         self,
     515         2884 :         key_end: Key,
     516         2884 :         ctx: &RequestContext,
     517         2884 :     ) -> anyhow::Result<(PersistentLayerDesc, Utf8PathBuf)> {
     518         2884 :         let temp_path = self.path.clone();
     519         2884 :         let result = self.finish0(key_end, ctx).await;
     520         2884 :         if let Err(ref e) = result {
     521            0 :             tracing::info!(%temp_path, "cleaning up temporary file after error during writing: {e}");
     522            0 :             if let Err(e) = std::fs::remove_file(&temp_path) {
     523            0 :                 tracing::warn!(error=%e, %temp_path, "error cleaning up temporary layer file after error during writing");
     524            0 :             }
     525         2884 :         }
     526         2884 :         result
     527         2884 :     }
     528              : 
     529         2884 :     async fn finish0(
     530         2884 :         self,
     531         2884 :         key_end: Key,
     532         2884 :         ctx: &RequestContext,
     533         2884 :     ) -> anyhow::Result<(PersistentLayerDesc, Utf8PathBuf)> {
     534         2884 :         let index_start_blk = self.blob_writer.size().div_ceil(PAGE_SZ as u64) as u32;
     535              : 
     536         2884 :         let mut file = self.blob_writer.into_inner(ctx).await?;
     537              : 
     538              :         // Write out the index
     539         2884 :         let (index_root_blk, block_buf) = self.tree.finish()?;
     540         2884 :         file.seek(SeekFrom::Start(index_start_blk as u64 * PAGE_SZ as u64))
     541         2884 :             .await?;
     542        30194 :         for buf in block_buf.blocks {
     543        27310 :             let (_buf, res) = file.write_all(buf.slice_len(), ctx).await;
     544        27310 :             res?;
     545              :         }
     546         2884 :         assert!(self.lsn_range.start < self.lsn_range.end);
     547              :         // Fill in the summary on blk 0
     548         2884 :         let summary = Summary {
     549         2884 :             magic: DELTA_FILE_MAGIC,
     550         2884 :             format_version: STORAGE_FORMAT_VERSION,
     551         2884 :             tenant_id: self.tenant_shard_id.tenant_id,
     552         2884 :             timeline_id: self.timeline_id,
     553         2884 :             key_range: self.key_start..key_end,
     554         2884 :             lsn_range: self.lsn_range.clone(),
     555         2884 :             index_start_blk,
     556         2884 :             index_root_blk,
     557         2884 :         };
     558         2884 : 
     559         2884 :         let mut buf = Vec::with_capacity(PAGE_SZ);
     560         2884 :         // TODO: could use smallvec here but it's a pain with Slice<T>
     561         2884 :         Summary::ser_into(&summary, &mut buf)?;
     562         2884 :         file.seek(SeekFrom::Start(0)).await?;
     563         2884 :         let (_buf, res) = file.write_all(buf.slice_len(), ctx).await;
     564         2884 :         res?;
     565              : 
     566         2884 :         let metadata = file
     567         2884 :             .metadata()
     568         2884 :             .await
     569         2884 :             .context("get file metadata to determine size")?;
     570              : 
     571              :         // 5GB limit for objects without multipart upload (which we don't want to use)
     572              :         // Make it a little bit below to account for differing GB units
     573              :         // https://docs.aws.amazon.com/AmazonS3/latest/userguide/upload-objects.html
     574         2884 :         ensure!(
     575         2884 :             metadata.len() <= S3_UPLOAD_LIMIT,
     576            0 :             "Created delta layer file at {} of size {} above limit {S3_UPLOAD_LIMIT}!",
     577            0 :             file.path(),
     578            0 :             metadata.len()
     579              :         );
     580              : 
     581              :         // Note: Because we opened the file in write-only mode, we cannot
     582              :         // reuse the same VirtualFile for reading later. That's why we don't
     583              :         // set inner.file here. The first read will have to re-open it.
     584              : 
     585         2884 :         let desc = PersistentLayerDesc::new_delta(
     586         2884 :             self.tenant_shard_id,
     587         2884 :             self.timeline_id,
     588         2884 :             self.key_start..key_end,
     589         2884 :             self.lsn_range.clone(),
     590         2884 :             metadata.len(),
     591         2884 :         );
     592         2884 : 
     593         2884 :         // fsync the file
     594         2884 :         file.sync_all()
     595         2884 :             .await
     596         2884 :             .maybe_fatal_err("delta_layer sync_all")?;
     597              : 
     598         2884 :         trace!("created delta layer {}", self.path);
     599              : 
     600         2884 :         Ok((desc, self.path))
     601         2884 :     }
     602              : }
     603              : 
     604              : /// A builder object for constructing a new delta layer.
     605              : ///
     606              : /// Usage:
     607              : ///
     608              : /// 1. Create the DeltaLayerWriter by calling DeltaLayerWriter::new(...)
     609              : ///
     610              : /// 2. Write the contents by calling `put_value` for every page
     611              : ///    version to store in the layer.
     612              : ///
     613              : /// 3. Call `finish`.
     614              : ///
     615              : /// # Note
     616              : ///
     617              : /// As described in <https://github.com/neondatabase/neon/issues/2650>, it's
     618              : /// possible for the writer to drop before `finish` is actually called. So this
     619              : /// could lead to odd temporary files in the directory, exhausting file system.
     620              : /// This structure wraps `DeltaLayerWriterInner` and also contains `Drop`
     621              : /// implementation that cleans up the temporary file in failure. It's not
     622              : /// possible to do this directly in `DeltaLayerWriterInner` since `finish` moves
     623              : /// out some fields, making it impossible to implement `Drop`.
     624              : ///
     625              : #[must_use]
     626              : pub struct DeltaLayerWriter {
     627              :     inner: Option<DeltaLayerWriterInner>,
     628              : }
     629              : 
     630              : impl DeltaLayerWriter {
     631              :     ///
     632              :     /// Start building a new delta layer.
     633              :     ///
     634         2936 :     pub async fn new(
     635         2936 :         conf: &'static PageServerConf,
     636         2936 :         timeline_id: TimelineId,
     637         2936 :         tenant_shard_id: TenantShardId,
     638         2936 :         key_start: Key,
     639         2936 :         lsn_range: Range<Lsn>,
     640         2936 :         ctx: &RequestContext,
     641         2936 :     ) -> anyhow::Result<Self> {
     642         2936 :         Ok(Self {
     643         2936 :             inner: Some(
     644         2936 :                 DeltaLayerWriterInner::new(
     645         2936 :                     conf,
     646         2936 :                     timeline_id,
     647         2936 :                     tenant_shard_id,
     648         2936 :                     key_start,
     649         2936 :                     lsn_range,
     650         2936 :                     ctx,
     651         2936 :                 )
     652         2936 :                 .await?,
     653              :             ),
     654              :         })
     655         2936 :     }
     656              : 
     657            0 :     pub fn is_empty(&self) -> bool {
     658            0 :         self.inner.as_ref().unwrap().num_keys == 0
     659            0 :     }
     660              : 
     661              :     ///
     662              :     /// Append a key-value pair to the file.
     663              :     ///
     664              :     /// The values must be appended in key, lsn order.
     665              :     ///
     666      4169428 :     pub async fn put_value(
     667      4169428 :         &mut self,
     668      4169428 :         key: Key,
     669      4169428 :         lsn: Lsn,
     670      4169428 :         val: Value,
     671      4169428 :         ctx: &RequestContext,
     672      4169428 :     ) -> anyhow::Result<()> {
     673      4169428 :         self.inner
     674      4169428 :             .as_mut()
     675      4169428 :             .unwrap()
     676      4169428 :             .put_value(key, lsn, val, ctx)
     677      4169428 :             .await
     678      4169428 :     }
     679              : 
     680      8773320 :     pub async fn put_value_bytes<Buf>(
     681      8773320 :         &mut self,
     682      8773320 :         key: Key,
     683      8773320 :         lsn: Lsn,
     684      8773320 :         val: FullSlice<Buf>,
     685      8773320 :         will_init: bool,
     686      8773320 :         ctx: &RequestContext,
     687      8773320 :     ) -> (FullSlice<Buf>, anyhow::Result<()>)
     688      8773320 :     where
     689      8773320 :         Buf: IoBuf + Send,
     690      8773320 :     {
     691      8773320 :         self.inner
     692      8773320 :             .as_mut()
     693      8773320 :             .unwrap()
     694      8773320 :             .put_value_bytes(key, lsn, val, will_init, ctx)
     695      8773320 :             .await
     696      8773320 :     }
     697              : 
     698      4047944 :     pub fn size(&self) -> u64 {
     699      4047944 :         self.inner.as_ref().unwrap().size()
     700      4047944 :     }
     701              : 
     702              :     ///
     703              :     /// Finish writing the delta layer.
     704              :     ///
     705         2884 :     pub(crate) async fn finish(
     706         2884 :         mut self,
     707         2884 :         key_end: Key,
     708         2884 :         ctx: &RequestContext,
     709         2884 :     ) -> anyhow::Result<(PersistentLayerDesc, Utf8PathBuf)> {
     710         2884 :         self.inner.take().unwrap().finish(key_end, ctx).await
     711         2884 :     }
     712              : 
     713        24508 :     pub(crate) fn num_keys(&self) -> usize {
     714        24508 :         self.inner.as_ref().unwrap().num_keys
     715        24508 :     }
     716              : 
     717        30280 :     pub(crate) fn estimated_size(&self) -> u64 {
     718        30280 :         let inner = self.inner.as_ref().unwrap();
     719        30280 :         inner.blob_writer.size() + inner.tree.borrow_writer().size() + PAGE_SZ as u64
     720        30280 :     }
     721              : }
     722              : 
     723              : impl Drop for DeltaLayerWriter {
     724         2936 :     fn drop(&mut self) {
     725         2936 :         if let Some(inner) = self.inner.take() {
     726           52 :             // We want to remove the virtual file here, so it's fine to not
     727           52 :             // having completely flushed unwritten data.
     728           52 :             let vfile = inner.blob_writer.into_inner_no_flush();
     729           52 :             vfile.remove();
     730         2884 :         }
     731         2936 :     }
     732              : }
     733              : 
     734              : #[derive(thiserror::Error, Debug)]
     735              : pub enum RewriteSummaryError {
     736              :     #[error("magic mismatch")]
     737              :     MagicMismatch,
     738              :     #[error(transparent)]
     739              :     Other(#[from] anyhow::Error),
     740              : }
     741              : 
     742              : impl From<std::io::Error> for RewriteSummaryError {
     743            0 :     fn from(e: std::io::Error) -> Self {
     744            0 :         Self::Other(anyhow::anyhow!(e))
     745            0 :     }
     746              : }
     747              : 
     748              : impl DeltaLayer {
     749            0 :     pub async fn rewrite_summary<F>(
     750            0 :         path: &Utf8Path,
     751            0 :         rewrite: F,
     752            0 :         ctx: &RequestContext,
     753            0 :     ) -> Result<(), RewriteSummaryError>
     754            0 :     where
     755            0 :         F: Fn(Summary) -> Summary,
     756            0 :     {
     757            0 :         let mut file = VirtualFile::open_with_options(
     758            0 :             path,
     759            0 :             virtual_file::OpenOptions::new().read(true).write(true),
     760            0 :             ctx,
     761            0 :         )
     762            0 :         .await
     763            0 :         .with_context(|| format!("Failed to open file '{}'", path))?;
     764            0 :         let file_id = page_cache::next_file_id();
     765            0 :         let block_reader = FileBlockReader::new(&file, file_id);
     766            0 :         let summary_blk = block_reader.read_blk(0, ctx).await?;
     767            0 :         let actual_summary = Summary::des_prefix(summary_blk.as_ref()).context("deserialize")?;
     768            0 :         if actual_summary.magic != DELTA_FILE_MAGIC {
     769            0 :             return Err(RewriteSummaryError::MagicMismatch);
     770            0 :         }
     771            0 : 
     772            0 :         let new_summary = rewrite(actual_summary);
     773            0 : 
     774            0 :         let mut buf = Vec::with_capacity(PAGE_SZ);
     775            0 :         // TODO: could use smallvec here, but it's a pain with Slice<T>
     776            0 :         Summary::ser_into(&new_summary, &mut buf).context("serialize")?;
     777            0 :         file.seek(SeekFrom::Start(0)).await?;
     778            0 :         let (_buf, res) = file.write_all(buf.slice_len(), ctx).await;
     779            0 :         res?;
     780            0 :         Ok(())
     781            0 :     }
     782              : }
     783              : 
     784              : impl DeltaLayerInner {
     785         2072 :     pub(crate) fn key_range(&self) -> &Range<Key> {
     786         2072 :         &self.layer_key_range
     787         2072 :     }
     788              : 
     789         2072 :     pub(crate) fn lsn_range(&self) -> &Range<Lsn> {
     790         2072 :         &self.layer_lsn_range
     791         2072 :     }
     792              : 
     793         2188 :     pub(super) async fn load(
     794         2188 :         path: &Utf8Path,
     795         2188 :         summary: Option<Summary>,
     796         2188 :         max_vectored_read_bytes: Option<MaxVectoredReadBytes>,
     797         2188 :         ctx: &RequestContext,
     798         2188 :     ) -> anyhow::Result<Self> {
     799         2188 :         let file = Arc::new(
     800         2188 :             VirtualFile::open_v2(path, ctx)
     801         2188 :                 .await
     802         2188 :                 .context("open layer file")?,
     803              :         );
     804              : 
     805         2188 :         let file_id = page_cache::next_file_id();
     806         2188 : 
     807         2188 :         let block_reader = FileBlockReader::new(&file, file_id);
     808              : 
     809         2188 :         let summary_blk = block_reader
     810         2188 :             .read_blk(0, ctx)
     811         2188 :             .await
     812         2188 :             .context("read first block")?;
     813              : 
     814              :         // TODO: this should be an assertion instead; see ImageLayerInner::load
     815         2188 :         let actual_summary =
     816         2188 :             Summary::des_prefix(summary_blk.as_ref()).context("deserialize first block")?;
     817              : 
     818         2188 :         if let Some(mut expected_summary) = summary {
     819              :             // production code path
     820         2188 :             expected_summary.index_start_blk = actual_summary.index_start_blk;
     821         2188 :             expected_summary.index_root_blk = actual_summary.index_root_blk;
     822         2188 :             // mask out the timeline_id, but still require the layers to be from the same tenant
     823         2188 :             expected_summary.timeline_id = actual_summary.timeline_id;
     824         2188 : 
     825         2188 :             if actual_summary != expected_summary {
     826            0 :                 bail!(
     827            0 :                     "in-file summary does not match expected summary. actual = {:?} expected = {:?}",
     828            0 :                     actual_summary,
     829            0 :                     expected_summary
     830            0 :                 );
     831         2188 :             }
     832            0 :         }
     833              : 
     834         2188 :         Ok(DeltaLayerInner {
     835         2188 :             file,
     836         2188 :             file_id,
     837         2188 :             index_start_blk: actual_summary.index_start_blk,
     838         2188 :             index_root_blk: actual_summary.index_root_blk,
     839         2188 :             max_vectored_read_bytes,
     840         2188 :             layer_key_range: actual_summary.key_range,
     841         2188 :             layer_lsn_range: actual_summary.lsn_range,
     842         2188 :         })
     843         2188 :     }
     844              : 
     845              :     // Look up the keys in the provided keyspace and update
     846              :     // the reconstruct state with whatever is found.
     847              :     //
     848              :     // Currently, the index is visited for each range, but this
     849              :     // can be further optimised to visit the index only once.
     850       434324 :     pub(super) async fn get_values_reconstruct_data(
     851       434324 :         &self,
     852       434324 :         this: ResidentLayer,
     853       434324 :         keyspace: KeySpace,
     854       434324 :         lsn_range: Range<Lsn>,
     855       434324 :         reconstruct_state: &mut ValuesReconstructState,
     856       434324 :         ctx: &RequestContext,
     857       434324 :     ) -> Result<(), GetVectoredError> {
     858       434324 :         let block_reader = FileBlockReader::new(&self.file, self.file_id);
     859       434324 :         let index_reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
     860       434324 :             self.index_start_blk,
     861       434324 :             self.index_root_blk,
     862       434324 :             block_reader,
     863       434324 :         );
     864       434324 : 
     865       434324 :         let planner = VectoredReadPlanner::new(
     866       434324 :             self.max_vectored_read_bytes
     867       434324 :                 .expect("Layer is loaded with max vectored bytes config")
     868       434324 :                 .0
     869       434324 :                 .into(),
     870       434324 :         );
     871       434324 : 
     872       434324 :         let data_end_offset = self.index_start_offset();
     873              : 
     874       434324 :         let reads = Self::plan_reads(
     875       434324 :             &keyspace,
     876       434324 :             lsn_range.clone(),
     877       434324 :             data_end_offset,
     878       434324 :             index_reader,
     879       434324 :             planner,
     880       434324 :             ctx,
     881       434324 :         )
     882       434324 :         .await
     883       434324 :         .map_err(GetVectoredError::Other)?;
     884              : 
     885       434324 :         self.do_reads_and_update_state(this, reads, reconstruct_state, ctx)
     886       434324 :             .await;
     887              : 
     888       434324 :         Ok(())
     889       434324 :     }
     890              : 
     891       434728 :     async fn plan_reads<Reader>(
     892       434728 :         keyspace: &KeySpace,
     893       434728 :         lsn_range: Range<Lsn>,
     894       434728 :         data_end_offset: u64,
     895       434728 :         index_reader: DiskBtreeReader<Reader, DELTA_KEY_SIZE>,
     896       434728 :         mut planner: VectoredReadPlanner,
     897       434728 :         ctx: &RequestContext,
     898       434728 :     ) -> anyhow::Result<Vec<VectoredRead>>
     899       434728 :     where
     900       434728 :         Reader: BlockReader + Clone,
     901       434728 :     {
     902       434728 :         let ctx = RequestContextBuilder::extend(ctx)
     903       434728 :             .page_content_kind(PageContentKind::DeltaLayerBtreeNode)
     904       434728 :             .build();
     905              : 
     906       435180 :         for range in keyspace.ranges.iter() {
     907       435180 :             let mut range_end_handled = false;
     908       435180 : 
     909       435180 :             let start_key = DeltaKey::from_key_lsn(&range.start, lsn_range.start);
     910       435180 :             let index_stream = index_reader.clone().into_stream(&start_key.0, &ctx);
     911       435180 :             let mut index_stream = std::pin::pin!(index_stream);
     912              : 
     913      1272171 :             while let Some(index_entry) = index_stream.next().await {
     914      1255527 :                 let (raw_key, value) = index_entry?;
     915      1255527 :                 let key = Key::from_slice(&raw_key[..KEY_SIZE]);
     916      1255527 :                 let lsn = DeltaKey::extract_lsn_from_buf(&raw_key);
     917      1255527 :                 let blob_ref = BlobRef(value);
     918      1255527 : 
     919      1255527 :                 // Lsns are not monotonically increasing across keys, so we don't assert on them.
     920      1255527 :                 assert!(key >= range.start);
     921              : 
     922      1255527 :                 let outside_lsn_range = !lsn_range.contains(&lsn);
     923              : 
     924      1255527 :                 let flag = {
     925      1255527 :                     if outside_lsn_range {
     926        84632 :                         BlobFlag::Ignore
     927      1170895 :                     } else if blob_ref.will_init() {
     928      1053643 :                         BlobFlag::ReplaceAll
     929              :                     } else {
     930              :                         // Usual path: add blob to the read
     931       117252 :                         BlobFlag::None
     932              :                     }
     933              :                 };
     934              : 
     935      1255527 :                 if key >= range.end || (key.next() == range.end && lsn >= lsn_range.end) {
     936       418536 :                     planner.handle_range_end(blob_ref.pos());
     937       418536 :                     range_end_handled = true;
     938       418536 :                     break;
     939       836991 :                 } else {
     940       836991 :                     planner.handle(key, lsn, blob_ref.pos(), flag);
     941       836991 :                 }
     942              :             }
     943              : 
     944       435180 :             if !range_end_handled {
     945        16644 :                 tracing::debug!("Handling range end fallback at {}", data_end_offset);
     946        16644 :                 planner.handle_range_end(data_end_offset);
     947       418536 :             }
     948              :         }
     949              : 
     950       434728 :         Ok(planner.finish())
     951       434728 :     }
     952              : 
     953       434724 :     fn get_min_read_buffer_size(
     954       434724 :         planned_reads: &[VectoredRead],
     955       434724 :         read_size_soft_max: usize,
     956       434724 :     ) -> usize {
     957       434724 :         let Some(largest_read) = planned_reads.iter().max_by_key(|read| read.size()) else {
     958       172929 :             return read_size_soft_max;
     959              :         };
     960              : 
     961       261795 :         let largest_read_size = largest_read.size();
     962       261795 :         if largest_read_size > read_size_soft_max {
     963              :             // If the read is oversized, it should only contain one key.
     964          400 :             let offenders = largest_read
     965          400 :                 .blobs_at
     966          400 :                 .as_slice()
     967          400 :                 .iter()
     968          400 :                 .filter_map(|(_, blob_meta)| {
     969          400 :                     if blob_meta.key.is_rel_dir_key()
     970          400 :                         || blob_meta.key == DBDIR_KEY
     971          400 :                         || blob_meta.key.is_aux_file_key()
     972              :                     {
     973              :                         // The size of values for these keys is unbounded and can
     974              :                         // grow very large in pathological cases.
     975            0 :                         None
     976              :                     } else {
     977          400 :                         Some(format!("{}@{}", blob_meta.key, blob_meta.lsn))
     978              :                     }
     979          400 :                 })
     980          400 :                 .join(", ");
     981          400 : 
     982          400 :             if !offenders.is_empty() {
     983          400 :                 tracing::warn!(
     984            0 :                     "Oversized vectored read ({} > {}) for keys {}",
     985              :                     largest_read_size,
     986              :                     read_size_soft_max,
     987              :                     offenders
     988              :                 );
     989            0 :             }
     990       261395 :         }
     991              : 
     992       261795 :         largest_read_size
     993       434724 :     }
     994              : 
     995       434324 :     async fn do_reads_and_update_state(
     996       434324 :         &self,
     997       434324 :         this: ResidentLayer,
     998       434324 :         reads: Vec<VectoredRead>,
     999       434324 :         reconstruct_state: &mut ValuesReconstructState,
    1000       434324 :         ctx: &RequestContext,
    1001       434324 :     ) {
    1002       434324 :         let max_vectored_read_bytes = self
    1003       434324 :             .max_vectored_read_bytes
    1004       434324 :             .expect("Layer is loaded with max vectored bytes config")
    1005       434324 :             .0
    1006       434324 :             .into();
    1007       434324 :         let buf_size = Self::get_min_read_buffer_size(&reads, max_vectored_read_bytes);
    1008              : 
    1009              :         // Note that reads are processed in reverse order (from highest key+lsn).
    1010              :         // This is the order that `ReconstructState` requires such that it can
    1011              :         // track when a key is done.
    1012       434324 :         for read in reads.into_iter().rev() {
    1013       267425 :             let mut ios: HashMap<(Key, Lsn), OnDiskValueIo> = Default::default();
    1014       374766 :             for (_, blob_meta) in read.blobs_at.as_slice().iter().rev() {
    1015       374766 :                 let io = reconstruct_state.update_key(
    1016       374766 :                     &blob_meta.key,
    1017       374766 :                     blob_meta.lsn,
    1018       374766 :                     blob_meta.will_init,
    1019       374766 :                 );
    1020       374766 :                 ios.insert((blob_meta.key, blob_meta.lsn), io);
    1021       374766 :             }
    1022              : 
    1023       267425 :             let read_extend_residency = this.clone();
    1024       267425 :             let read_from = self.file.clone();
    1025       267425 :             let read_ctx = ctx.attached_child();
    1026       267425 :             reconstruct_state
    1027       267425 :                 .spawn_io(async move {
    1028       267425 :                     let vectored_blob_reader = VectoredBlobReader::new(&read_from);
    1029       267425 :                     let buf = IoBufferMut::with_capacity(buf_size);
    1030              : 
    1031       267425 :                     let res = vectored_blob_reader.read_blobs(&read, buf, &read_ctx).await;
    1032       267425 :                     match res {
    1033       267425 :                         Ok(blobs_buf) => {
    1034       267425 :                             let view = BufView::new_slice(&blobs_buf.buf);
    1035       374766 :                             for meta in blobs_buf.blobs.iter().rev() {
    1036       374766 :                                 let io = ios.remove(&(meta.meta.key, meta.meta.lsn)).unwrap();
    1037              : 
    1038       374766 :                                 let blob_read = meta.read(&view).await;
    1039       374766 :                                 let blob_read = match blob_read {
    1040       374766 :                                     Ok(buf) => buf,
    1041            0 :                                     Err(e) => {
    1042            0 :                                         io.complete(Err(e));
    1043            0 :                                         continue;
    1044              :                                     }
    1045              :                                 };
    1046              : 
    1047       374766 :                                 io.complete(Ok(OnDiskValue::WalRecordOrImage(
    1048       374766 :                                     blob_read.into_bytes(),
    1049       374766 :                                 )));
    1050              :                             }
    1051              : 
    1052       267425 :                             assert!(ios.is_empty());
    1053              :                         }
    1054            0 :                         Err(err) => {
    1055            0 :                             for (_, sender) in ios {
    1056            0 :                                 sender.complete(Err(std::io::Error::new(
    1057            0 :                                     err.kind(),
    1058            0 :                                     "vec read failed",
    1059            0 :                                 )));
    1060            0 :                             }
    1061              :                         }
    1062              :                     }
    1063              : 
    1064              :                     // keep layer resident until this IO is done; this spawned IO future generally outlives the
    1065              :                     // call to `self` / the `Arc<DownloadedLayer>` / the `ResidentLayer` that guarantees residency
    1066       267425 :                     drop(read_extend_residency);
    1067       267425 :                 })
    1068       267425 :                 .await;
    1069              :         }
    1070       434324 :     }
    1071              : 
    1072          812 :     pub(crate) async fn index_entries<'a>(
    1073          812 :         &'a self,
    1074          812 :         ctx: &RequestContext,
    1075          812 :     ) -> Result<Vec<DeltaEntry<'a>>> {
    1076          812 :         let block_reader = FileBlockReader::new(&self.file, self.file_id);
    1077          812 :         let tree_reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
    1078          812 :             self.index_start_blk,
    1079          812 :             self.index_root_blk,
    1080          812 :             block_reader,
    1081          812 :         );
    1082          812 : 
    1083          812 :         let mut all_keys: Vec<DeltaEntry<'_>> = Vec::new();
    1084          812 : 
    1085          812 :         tree_reader
    1086          812 :             .visit(
    1087          812 :                 &[0u8; DELTA_KEY_SIZE],
    1088          812 :                 VisitDirection::Forwards,
    1089      4128092 :                 |key, value| {
    1090      4128092 :                     let delta_key = DeltaKey::from_slice(key);
    1091      4128092 :                     let val_ref = ValueRef {
    1092      4128092 :                         blob_ref: BlobRef(value),
    1093      4128092 :                         layer: self,
    1094      4128092 :                     };
    1095      4128092 :                     let pos = BlobRef(value).pos();
    1096      4128092 :                     if let Some(last) = all_keys.last_mut() {
    1097      4127280 :                         // subtract offset of the current and last entries to get the size
    1098      4127280 :                         // of the value associated with this (key, lsn) tuple
    1099      4127280 :                         let first_pos = last.size;
    1100      4127280 :                         last.size = pos - first_pos;
    1101      4127280 :                     }
    1102      4128092 :                     let entry = DeltaEntry {
    1103      4128092 :                         key: delta_key.key(),
    1104      4128092 :                         lsn: delta_key.lsn(),
    1105      4128092 :                         size: pos,
    1106      4128092 :                         val: val_ref,
    1107      4128092 :                     };
    1108      4128092 :                     all_keys.push(entry);
    1109      4128092 :                     true
    1110      4128092 :                 },
    1111          812 :                 &RequestContextBuilder::extend(ctx)
    1112          812 :                     .page_content_kind(PageContentKind::DeltaLayerBtreeNode)
    1113          812 :                     .build(),
    1114          812 :             )
    1115          812 :             .await?;
    1116          812 :         if let Some(last) = all_keys.last_mut() {
    1117          812 :             // Last key occupies all space till end of value storage,
    1118          812 :             // which corresponds to beginning of the index
    1119          812 :             last.size = self.index_start_offset() - last.size;
    1120          812 :         }
    1121          812 :         Ok(all_keys)
    1122          812 :     }
    1123              : 
    1124              :     /// Using the given writer, write out a version which has the earlier Lsns than `until`.
    1125              :     ///
    1126              :     /// Return the amount of key value records pushed to the writer.
    1127           20 :     pub(super) async fn copy_prefix(
    1128           20 :         &self,
    1129           20 :         writer: &mut DeltaLayerWriter,
    1130           20 :         until: Lsn,
    1131           20 :         ctx: &RequestContext,
    1132           20 :     ) -> anyhow::Result<usize> {
    1133              :         use crate::tenant::vectored_blob_io::{
    1134              :             BlobMeta, ChunkedVectoredReadBuilder, VectoredReadExtended,
    1135              :         };
    1136              :         use futures::stream::TryStreamExt;
    1137              : 
    1138              :         #[derive(Debug)]
    1139              :         enum Item {
    1140              :             Actual(Key, Lsn, BlobRef),
    1141              :             Sentinel,
    1142              :         }
    1143              : 
    1144              :         impl From<Item> for Option<(Key, Lsn, BlobRef)> {
    1145          140 :             fn from(value: Item) -> Self {
    1146          140 :                 match value {
    1147          120 :                     Item::Actual(key, lsn, blob) => Some((key, lsn, blob)),
    1148           20 :                     Item::Sentinel => None,
    1149              :                 }
    1150          140 :             }
    1151              :         }
    1152              : 
    1153              :         impl Item {
    1154          140 :             fn offset(&self) -> Option<BlobRef> {
    1155          140 :                 match self {
    1156          120 :                     Item::Actual(_, _, blob) => Some(*blob),
    1157           20 :                     Item::Sentinel => None,
    1158              :                 }
    1159          140 :             }
    1160              : 
    1161          140 :             fn is_last(&self) -> bool {
    1162          140 :                 matches!(self, Item::Sentinel)
    1163          140 :             }
    1164              :         }
    1165              : 
    1166           20 :         let block_reader = FileBlockReader::new(&self.file, self.file_id);
    1167           20 :         let tree_reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
    1168           20 :             self.index_start_blk,
    1169           20 :             self.index_root_blk,
    1170           20 :             block_reader,
    1171           20 :         );
    1172           20 : 
    1173           20 :         let stream = self.stream_index_forwards(tree_reader, &[0u8; DELTA_KEY_SIZE], ctx);
    1174          120 :         let stream = stream.map_ok(|(key, lsn, pos)| Item::Actual(key, lsn, pos));
    1175           20 :         // put in a sentinel value for getting the end offset for last item, and not having to
    1176           20 :         // repeat the whole read part
    1177           20 :         let stream = stream.chain(futures::stream::once(futures::future::ready(Ok(
    1178           20 :             Item::Sentinel,
    1179           20 :         ))));
    1180           20 :         let mut stream = std::pin::pin!(stream);
    1181           20 : 
    1182           20 :         let mut prev: Option<(Key, Lsn, BlobRef)> = None;
    1183           20 : 
    1184           20 :         let mut read_builder: Option<ChunkedVectoredReadBuilder> = None;
    1185           20 : 
    1186           20 :         let max_read_size = self
    1187           20 :             .max_vectored_read_bytes
    1188           20 :             .map(|x| x.0.get())
    1189           20 :             .unwrap_or(8192);
    1190           20 : 
    1191           20 :         let mut buffer = Some(IoBufferMut::with_capacity(max_read_size));
    1192           20 : 
    1193           20 :         // FIXME: buffering of DeltaLayerWriter
    1194           20 :         let mut per_blob_copy = Vec::new();
    1195           20 : 
    1196           20 :         let mut records = 0;
    1197              : 
    1198          160 :         while let Some(item) = stream.try_next().await? {
    1199          140 :             tracing::debug!(?item, "popped");
    1200          140 :             let offset = item
    1201          140 :                 .offset()
    1202          140 :                 .unwrap_or(BlobRef::new(self.index_start_offset(), false));
    1203              : 
    1204          140 :             let actionable = if let Some((key, lsn, start_offset)) = prev.take() {
    1205          120 :                 let end_offset = offset;
    1206          120 : 
    1207          120 :                 Some((
    1208          120 :                     BlobMeta {
    1209          120 :                         key,
    1210          120 :                         lsn,
    1211          120 :                         will_init: false,
    1212          120 :                     },
    1213          120 :                     start_offset..end_offset,
    1214          120 :                 ))
    1215              :             } else {
    1216           20 :                 None
    1217              :             };
    1218              : 
    1219          140 :             let is_last = item.is_last();
    1220          140 : 
    1221          140 :             prev = Option::from(item);
    1222          140 : 
    1223          140 :             let actionable = actionable.filter(|x| x.0.lsn < until);
    1224              : 
    1225          140 :             let builder = if let Some((meta, offsets)) = actionable {
    1226              :                 // extend or create a new builder
    1227           64 :                 if read_builder
    1228           64 :                     .as_mut()
    1229           64 :                     .map(|x| x.extend(offsets.start.pos(), offsets.end.pos(), meta))
    1230           64 :                     .unwrap_or(VectoredReadExtended::No)
    1231           64 :                     == VectoredReadExtended::Yes
    1232              :                 {
    1233           32 :                     None
    1234              :                 } else {
    1235           32 :                     read_builder.replace(ChunkedVectoredReadBuilder::new(
    1236           32 :                         offsets.start.pos(),
    1237           32 :                         offsets.end.pos(),
    1238           32 :                         meta,
    1239           32 :                         max_read_size,
    1240           32 :                     ))
    1241              :                 }
    1242              :             } else {
    1243              :                 // nothing to do, except perhaps flush any existing for the last element
    1244           76 :                 None
    1245              :             };
    1246              : 
    1247              :             // flush the possible older builder and also the new one if the item was the last one
    1248          140 :             let builders = builder.into_iter();
    1249          140 :             let builders = if is_last {
    1250           20 :                 builders.chain(read_builder.take())
    1251              :             } else {
    1252          120 :                 builders.chain(None)
    1253              :             };
    1254              : 
    1255          172 :             for builder in builders {
    1256           32 :                 let read = builder.build();
    1257           32 : 
    1258           32 :                 let reader = VectoredBlobReader::new(&self.file);
    1259           32 : 
    1260           32 :                 let mut buf = buffer.take().unwrap();
    1261           32 : 
    1262           32 :                 buf.clear();
    1263           32 :                 buf.reserve(read.size());
    1264           32 :                 let res = reader.read_blobs(&read, buf, ctx).await?;
    1265              : 
    1266           32 :                 let view = BufView::new_slice(&res.buf);
    1267              : 
    1268           96 :                 for blob in res.blobs {
    1269           64 :                     let key = blob.meta.key;
    1270           64 :                     let lsn = blob.meta.lsn;
    1271              : 
    1272           64 :                     let data = blob.read(&view).await?;
    1273              : 
    1274              :                     #[cfg(debug_assertions)]
    1275           64 :                     Value::des(&data)
    1276           64 :                         .with_context(|| {
    1277            0 :                             format!(
    1278            0 :                                 "blob failed to deserialize for {}: {:?}",
    1279            0 :                                 blob,
    1280            0 :                                 utils::Hex(&data)
    1281            0 :                             )
    1282           64 :                         })
    1283           64 :                         .unwrap();
    1284           64 : 
    1285           64 :                     // is it an image or will_init walrecord?
    1286           64 :                     // FIXME: this could be handled by threading the BlobRef to the
    1287           64 :                     // VectoredReadBuilder
    1288           64 :                     let will_init = pageserver_api::value::ValueBytes::will_init(&data)
    1289           64 :                         .inspect_err(|_e| {
    1290            0 :                             #[cfg(feature = "testing")]
    1291            0 :                             tracing::error!(data=?utils::Hex(&data), err=?_e, %key, %lsn, "failed to parse will_init out of serialized value");
    1292           64 :                         })
    1293           64 :                         .unwrap_or(false);
    1294           64 : 
    1295           64 :                     per_blob_copy.clear();
    1296           64 :                     per_blob_copy.extend_from_slice(&data);
    1297              : 
    1298           64 :                     let (tmp, res) = writer
    1299           64 :                         .put_value_bytes(
    1300           64 :                             key,
    1301           64 :                             lsn,
    1302           64 :                             std::mem::take(&mut per_blob_copy).slice_len(),
    1303           64 :                             will_init,
    1304           64 :                             ctx,
    1305           64 :                         )
    1306           64 :                         .await;
    1307           64 :                     per_blob_copy = tmp.into_raw_slice().into_inner();
    1308           64 : 
    1309           64 :                     res?;
    1310              : 
    1311           64 :                     records += 1;
    1312              :                 }
    1313              : 
    1314           32 :                 buffer = Some(res.buf);
    1315              :             }
    1316              :         }
    1317              : 
    1318           20 :         assert!(
    1319           20 :             read_builder.is_none(),
    1320            0 :             "with the sentinel above loop should had handled all"
    1321              :         );
    1322              : 
    1323           20 :         Ok(records)
    1324           20 :     }
    1325              : 
    1326            8 :     pub(super) async fn dump(&self, ctx: &RequestContext) -> anyhow::Result<()> {
    1327            8 :         println!(
    1328            8 :             "index_start_blk: {}, root {}",
    1329            8 :             self.index_start_blk, self.index_root_blk
    1330            8 :         );
    1331            8 : 
    1332            8 :         let block_reader = FileBlockReader::new(&self.file, self.file_id);
    1333            8 :         let tree_reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
    1334            8 :             self.index_start_blk,
    1335            8 :             self.index_root_blk,
    1336            8 :             block_reader,
    1337            8 :         );
    1338            8 : 
    1339            8 :         tree_reader.dump().await?;
    1340              : 
    1341            8 :         let keys = self.index_entries(ctx).await?;
    1342              : 
    1343           16 :         async fn dump_blob(val: &ValueRef<'_>, ctx: &RequestContext) -> anyhow::Result<String> {
    1344           16 :             let buf = val.load_raw(ctx).await?;
    1345           16 :             let val = Value::des(&buf)?;
    1346           16 :             let desc = match val {
    1347           16 :                 Value::Image(img) => {
    1348           16 :                     format!(" img {} bytes", img.len())
    1349              :                 }
    1350            0 :                 Value::WalRecord(rec) => {
    1351            0 :                     let wal_desc = pageserver_api::record::describe_wal_record(&rec)?;
    1352            0 :                     format!(
    1353            0 :                         " rec {} bytes will_init: {} {}",
    1354            0 :                         buf.len(),
    1355            0 :                         rec.will_init(),
    1356            0 :                         wal_desc
    1357            0 :                     )
    1358              :                 }
    1359              :             };
    1360           16 :             Ok(desc)
    1361           16 :         }
    1362              : 
    1363           24 :         for entry in keys {
    1364           16 :             let DeltaEntry { key, lsn, val, .. } = entry;
    1365           16 :             let desc = match dump_blob(&val, ctx).await {
    1366           16 :                 Ok(desc) => desc,
    1367            0 :                 Err(err) => {
    1368            0 :                     format!("ERROR: {err}")
    1369              :                 }
    1370              :             };
    1371           16 :             println!("  key {key} at {lsn}: {desc}");
    1372              : 
    1373              :             // Print more details about CHECKPOINT records. Would be nice to print details
    1374              :             // of many other record types too, but these are particularly interesting, as
    1375              :             // have a lot of special processing for them in walingest.rs.
    1376            8 :             use pageserver_api::key::CHECKPOINT_KEY;
    1377            8 :             use postgres_ffi::CheckPoint;
    1378           16 :             if key == CHECKPOINT_KEY {
    1379            0 :                 let val = val.load(ctx).await?;
    1380            0 :                 match val {
    1381            0 :                     Value::Image(img) => {
    1382            0 :                         let checkpoint = CheckPoint::decode(&img)?;
    1383            0 :                         println!("   CHECKPOINT: {:?}", checkpoint);
    1384              :                     }
    1385            0 :                     Value::WalRecord(_rec) => {
    1386            0 :                         println!("   unexpected walrecord value for checkpoint key");
    1387            0 :                     }
    1388              :                 }
    1389           16 :             }
    1390              :         }
    1391              : 
    1392            8 :         Ok(())
    1393            8 :     }
    1394              : 
    1395           60 :     fn stream_index_forwards<'a, R>(
    1396           60 :         &'a self,
    1397           60 :         reader: DiskBtreeReader<R, DELTA_KEY_SIZE>,
    1398           60 :         start: &'a [u8; DELTA_KEY_SIZE],
    1399           60 :         ctx: &'a RequestContext,
    1400           60 :     ) -> impl futures::stream::Stream<
    1401           60 :         Item = Result<(Key, Lsn, BlobRef), crate::tenant::disk_btree::DiskBtreeError>,
    1402           60 :     > + 'a
    1403           60 :     where
    1404           60 :         R: BlockReader + 'a,
    1405           60 :     {
    1406              :         use futures::stream::TryStreamExt;
    1407           60 :         let stream = reader.into_stream(start, ctx);
    1408          304 :         stream.map_ok(|(key, value)| {
    1409          304 :             let key = DeltaKey::from_slice(&key);
    1410          304 :             let (key, lsn) = (key.key(), key.lsn());
    1411          304 :             let offset = BlobRef(value);
    1412          304 : 
    1413          304 :             (key, lsn, offset)
    1414          304 :         })
    1415           60 :     }
    1416              : 
    1417              :     /// The file offset to the first block of index.
    1418              :     ///
    1419              :     /// The file structure is summary, values, and index. We often need this for the size of last blob.
    1420       436368 :     fn index_start_offset(&self) -> u64 {
    1421       436368 :         let offset = self.index_start_blk as u64 * PAGE_SZ as u64;
    1422       436368 :         let bref = BlobRef(offset);
    1423       436368 :         tracing::debug!(
    1424              :             index_start_blk = self.index_start_blk,
    1425              :             offset,
    1426            0 :             pos = bref.pos(),
    1427            0 :             "index_start_offset"
    1428              :         );
    1429       436368 :         offset
    1430       436368 :     }
    1431              : 
    1432         1148 :     pub fn iter<'a>(&'a self, ctx: &'a RequestContext) -> DeltaLayerIterator<'a> {
    1433         1148 :         let block_reader = FileBlockReader::new(&self.file, self.file_id);
    1434         1148 :         let tree_reader =
    1435         1148 :             DiskBtreeReader::new(self.index_start_blk, self.index_root_blk, block_reader);
    1436         1148 :         DeltaLayerIterator {
    1437         1148 :             delta_layer: self,
    1438         1148 :             ctx,
    1439         1148 :             index_iter: tree_reader.iter(&[0; DELTA_KEY_SIZE], ctx),
    1440         1148 :             key_values_batch: std::collections::VecDeque::new(),
    1441         1148 :             is_end: false,
    1442         1148 :             planner: StreamingVectoredReadPlanner::new(
    1443         1148 :                 1024 * 8192, // The default value. Unit tests might use a different value. 1024 * 8K = 8MB buffer.
    1444         1148 :                 1024,        // The default value. Unit tests might use a different value
    1445         1148 :             ),
    1446         1148 :         }
    1447         1148 :     }
    1448              : 
    1449              :     /// NB: not super efficient, but not terrible either. Should prob be an iterator.
    1450              :     //
    1451              :     // We're reusing the index traversal logical in plan_reads; would be nice to
    1452              :     // factor that out.
    1453            0 :     pub(crate) async fn load_keys(&self, ctx: &RequestContext) -> anyhow::Result<Vec<Key>> {
    1454            0 :         self.index_entries(ctx)
    1455            0 :             .await
    1456            0 :             .map(|entries| entries.into_iter().map(|entry| entry.key).collect())
    1457            0 :     }
    1458              : }
    1459              : 
    1460              : /// A set of data associated with a delta layer key and its value
    1461              : pub struct DeltaEntry<'a> {
    1462              :     pub key: Key,
    1463              :     pub lsn: Lsn,
    1464              :     /// Size of the stored value
    1465              :     pub size: u64,
    1466              :     /// Reference to the on-disk value
    1467              :     pub val: ValueRef<'a>,
    1468              : }
    1469              : 
    1470              : /// Reference to an on-disk value
    1471              : pub struct ValueRef<'a> {
    1472              :     blob_ref: BlobRef,
    1473              :     layer: &'a DeltaLayerInner,
    1474              : }
    1475              : 
    1476              : impl ValueRef<'_> {
    1477              :     /// Loads the value from disk
    1478            0 :     pub async fn load(&self, ctx: &RequestContext) -> Result<Value> {
    1479            0 :         let buf = self.load_raw(ctx).await?;
    1480            0 :         let val = Value::des(&buf)?;
    1481            0 :         Ok(val)
    1482            0 :     }
    1483              : 
    1484           16 :     async fn load_raw(&self, ctx: &RequestContext) -> Result<Vec<u8>> {
    1485           16 :         let reader = BlockCursor::new(crate::tenant::block_io::BlockReaderRef::Adapter(Adapter(
    1486           16 :             self.layer,
    1487           16 :         )));
    1488           16 :         let buf = reader.read_blob(self.blob_ref.pos(), ctx).await?;
    1489           16 :         Ok(buf)
    1490           16 :     }
    1491              : }
    1492              : 
    1493              : pub(crate) struct Adapter<T>(T);
    1494              : 
    1495              : impl<T: AsRef<DeltaLayerInner>> Adapter<T> {
    1496           16 :     pub(crate) async fn read_blk(
    1497           16 :         &self,
    1498           16 :         blknum: u32,
    1499           16 :         ctx: &RequestContext,
    1500           16 :     ) -> Result<BlockLease, std::io::Error> {
    1501           16 :         let block_reader = FileBlockReader::new(&self.0.as_ref().file, self.0.as_ref().file_id);
    1502           16 :         block_reader.read_blk(blknum, ctx).await
    1503           16 :     }
    1504              : }
    1505              : 
    1506              : impl AsRef<DeltaLayerInner> for DeltaLayerInner {
    1507           32 :     fn as_ref(&self) -> &DeltaLayerInner {
    1508           32 :         self
    1509           32 :     }
    1510              : }
    1511              : 
    1512              : impl<'a> pageserver_compaction::interface::CompactionDeltaEntry<'a, Key> for DeltaEntry<'a> {
    1513            0 :     fn key(&self) -> Key {
    1514            0 :         self.key
    1515            0 :     }
    1516            0 :     fn lsn(&self) -> Lsn {
    1517            0 :         self.lsn
    1518            0 :     }
    1519            0 :     fn size(&self) -> u64 {
    1520            0 :         self.size
    1521            0 :     }
    1522              : }
    1523              : 
    1524              : pub struct DeltaLayerIterator<'a> {
    1525              :     delta_layer: &'a DeltaLayerInner,
    1526              :     ctx: &'a RequestContext,
    1527              :     planner: StreamingVectoredReadPlanner,
    1528              :     index_iter: DiskBtreeIterator<'a>,
    1529              :     key_values_batch: VecDeque<(Key, Lsn, Value)>,
    1530              :     is_end: bool,
    1531              : }
    1532              : 
    1533              : impl DeltaLayerIterator<'_> {
    1534            0 :     pub(crate) fn layer_dbg_info(&self) -> String {
    1535            0 :         self.delta_layer.layer_dbg_info()
    1536            0 :     }
    1537              : 
    1538              :     /// Retrieve a batch of key-value pairs into the iterator buffer.
    1539        42808 :     async fn next_batch(&mut self) -> anyhow::Result<()> {
    1540        42808 :         assert!(self.key_values_batch.is_empty());
    1541        42808 :         assert!(!self.is_end);
    1542              : 
    1543        42808 :         let plan = loop {
    1544      4199244 :             if let Some(res) = self.index_iter.next().await {
    1545      4198152 :                 let (raw_key, value) = res?;
    1546      4198152 :                 let key = Key::from_slice(&raw_key[..KEY_SIZE]);
    1547      4198152 :                 let lsn = DeltaKey::extract_lsn_from_buf(&raw_key);
    1548      4198152 :                 let blob_ref = BlobRef(value);
    1549      4198152 :                 let offset = blob_ref.pos();
    1550        41716 :                 if let Some(batch_plan) =
    1551      4198152 :                     self.planner.handle(key, lsn, offset, blob_ref.will_init())
    1552              :                 {
    1553        41716 :                     break batch_plan;
    1554      4156436 :                 }
    1555              :             } else {
    1556         1092 :                 self.is_end = true;
    1557         1092 :                 let data_end_offset = self.delta_layer.index_start_offset();
    1558         1092 :                 if let Some(item) = self.planner.handle_range_end(data_end_offset) {
    1559         1092 :                     break item;
    1560              :                 } else {
    1561            0 :                     return Ok(()); // TODO: test empty iterator
    1562              :                 }
    1563              :             }
    1564              :         };
    1565        42808 :         let vectored_blob_reader = VectoredBlobReader::new(&self.delta_layer.file);
    1566        42808 :         let mut next_batch = std::collections::VecDeque::new();
    1567        42808 :         let buf_size = plan.size();
    1568        42808 :         let buf = IoBufferMut::with_capacity(buf_size);
    1569        42808 :         let blobs_buf = vectored_blob_reader
    1570        42808 :             .read_blobs(&plan, buf, self.ctx)
    1571        42808 :             .await?;
    1572        42808 :         let view = BufView::new_slice(&blobs_buf.buf);
    1573      4198096 :         for meta in blobs_buf.blobs.iter() {
    1574      4198096 :             let blob_read = meta.read(&view).await?;
    1575      4198096 :             let value = Value::des(&blob_read)?;
    1576              : 
    1577      4198096 :             next_batch.push_back((meta.meta.key, meta.meta.lsn, value));
    1578              :         }
    1579        42808 :         self.key_values_batch = next_batch;
    1580        42808 :         Ok(())
    1581        42808 :     }
    1582              : 
    1583      4199668 :     pub async fn next(&mut self) -> anyhow::Result<Option<(Key, Lsn, Value)>> {
    1584      4199668 :         if self.key_values_batch.is_empty() {
    1585        44760 :             if self.is_end {
    1586         2120 :                 return Ok(None);
    1587        42640 :             }
    1588        42640 :             self.next_batch().await?;
    1589      4154908 :         }
    1590      4197548 :         Ok(Some(
    1591      4197548 :             self.key_values_batch
    1592      4197548 :                 .pop_front()
    1593      4197548 :                 .expect("should not be empty"),
    1594      4197548 :         ))
    1595      4199668 :     }
    1596              : }
    1597              : 
    1598              : #[cfg(test)]
    1599              : pub(crate) mod test {
    1600              :     use std::collections::BTreeMap;
    1601              : 
    1602              :     use itertools::MinMaxResult;
    1603              :     use rand::prelude::{SeedableRng, SliceRandom, StdRng};
    1604              :     use rand::RngCore;
    1605              : 
    1606              :     use super::*;
    1607              :     use crate::tenant::harness::TIMELINE_ID;
    1608              :     use crate::tenant::storage_layer::{Layer, ResidentLayer};
    1609              :     use crate::tenant::vectored_blob_io::StreamingVectoredReadPlanner;
    1610              :     use crate::tenant::{Tenant, Timeline};
    1611              :     use crate::{
    1612              :         context::DownloadBehavior,
    1613              :         task_mgr::TaskKind,
    1614              :         tenant::{disk_btree::tests::TestDisk, harness::TenantHarness},
    1615              :         DEFAULT_PG_VERSION,
    1616              :     };
    1617              :     use bytes::Bytes;
    1618              :     use pageserver_api::value::Value;
    1619              : 
    1620              :     /// Construct an index for a fictional delta layer and and then
    1621              :     /// traverse in order to plan vectored reads for a query. Finally,
    1622              :     /// verify that the traversal fed the right index key and value
    1623              :     /// pairs into the planner.
    1624              :     #[tokio::test]
    1625            4 :     async fn test_delta_layer_index_traversal() {
    1626            4 :         let base_key = Key {
    1627            4 :             field1: 0,
    1628            4 :             field2: 1663,
    1629            4 :             field3: 12972,
    1630            4 :             field4: 16396,
    1631            4 :             field5: 0,
    1632            4 :             field6: 246080,
    1633            4 :         };
    1634            4 : 
    1635            4 :         // Populate the index with some entries
    1636            4 :         let entries: BTreeMap<Key, Vec<Lsn>> = BTreeMap::from([
    1637            4 :             (base_key, vec![Lsn(1), Lsn(5), Lsn(25), Lsn(26), Lsn(28)]),
    1638            4 :             (base_key.add(1), vec![Lsn(2), Lsn(5), Lsn(10), Lsn(50)]),
    1639            4 :             (base_key.add(2), vec![Lsn(2), Lsn(5), Lsn(10), Lsn(50)]),
    1640            4 :             (base_key.add(5), vec![Lsn(10), Lsn(15), Lsn(16), Lsn(20)]),
    1641            4 :         ]);
    1642            4 : 
    1643            4 :         let mut disk = TestDisk::default();
    1644            4 :         let mut writer = DiskBtreeBuilder::<_, DELTA_KEY_SIZE>::new(&mut disk);
    1645            4 : 
    1646            4 :         let mut disk_offset = 0;
    1647           20 :         for (key, lsns) in &entries {
    1648           84 :             for lsn in lsns {
    1649           68 :                 let index_key = DeltaKey::from_key_lsn(key, *lsn);
    1650           68 :                 let blob_ref = BlobRef::new(disk_offset, false);
    1651           68 :                 writer
    1652           68 :                     .append(&index_key.0, blob_ref.0)
    1653           68 :                     .expect("In memory disk append should never fail");
    1654           68 : 
    1655           68 :                 disk_offset += 1;
    1656           68 :             }
    1657            4 :         }
    1658            4 : 
    1659            4 :         // Prepare all the arguments for the call into `plan_reads` below
    1660            4 :         let (root_offset, _writer) = writer
    1661            4 :             .finish()
    1662            4 :             .expect("In memory disk finish should never fail");
    1663            4 :         let reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(0, root_offset, disk);
    1664            4 :         let planner = VectoredReadPlanner::new(100);
    1665            4 :         let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error);
    1666            4 : 
    1667            4 :         let keyspace = KeySpace {
    1668            4 :             ranges: vec![
    1669            4 :                 base_key..base_key.add(3),
    1670            4 :                 base_key.add(3)..base_key.add(100),
    1671            4 :             ],
    1672            4 :         };
    1673            4 :         let lsn_range = Lsn(2)..Lsn(40);
    1674            4 : 
    1675            4 :         // Plan and validate
    1676            4 :         let vectored_reads = DeltaLayerInner::plan_reads(
    1677            4 :             &keyspace,
    1678            4 :             lsn_range.clone(),
    1679            4 :             disk_offset,
    1680            4 :             reader,
    1681            4 :             planner,
    1682            4 :             &ctx,
    1683            4 :         )
    1684            4 :         .await
    1685            4 :         .expect("Read planning should not fail");
    1686            4 : 
    1687            4 :         validate(keyspace, lsn_range, vectored_reads, entries);
    1688            4 :     }
    1689              : 
    1690            4 :     fn validate(
    1691            4 :         keyspace: KeySpace,
    1692            4 :         lsn_range: Range<Lsn>,
    1693            4 :         vectored_reads: Vec<VectoredRead>,
    1694            4 :         index_entries: BTreeMap<Key, Vec<Lsn>>,
    1695            4 :     ) {
    1696              :         #[derive(Debug, PartialEq, Eq)]
    1697              :         struct BlobSpec {
    1698              :             key: Key,
    1699              :             lsn: Lsn,
    1700              :             at: u64,
    1701              :         }
    1702              : 
    1703            4 :         let mut planned_blobs = Vec::new();
    1704           60 :         for read in vectored_reads {
    1705           56 :             for (at, meta) in read.blobs_at.as_slice() {
    1706           56 :                 planned_blobs.push(BlobSpec {
    1707           56 :                     key: meta.key,
    1708           56 :                     lsn: meta.lsn,
    1709           56 :                     at: *at,
    1710           56 :                 });
    1711           56 :             }
    1712              :         }
    1713              : 
    1714            4 :         let mut expected_blobs = Vec::new();
    1715            4 :         let mut disk_offset = 0;
    1716           20 :         for (key, lsns) in index_entries {
    1717           84 :             for lsn in lsns {
    1718           84 :                 let key_included = keyspace.ranges.iter().any(|range| range.contains(&key));
    1719           68 :                 let lsn_included = lsn_range.contains(&lsn);
    1720           68 : 
    1721           68 :                 if key_included && lsn_included {
    1722           56 :                     expected_blobs.push(BlobSpec {
    1723           56 :                         key,
    1724           56 :                         lsn,
    1725           56 :                         at: disk_offset,
    1726           56 :                     });
    1727           56 :                 }
    1728              : 
    1729           68 :                 disk_offset += 1;
    1730              :             }
    1731              :         }
    1732              : 
    1733            4 :         assert_eq!(planned_blobs, expected_blobs);
    1734            4 :     }
    1735              : 
    1736              :     mod constants {
    1737              :         use utils::lsn::Lsn;
    1738              : 
    1739              :         /// Offset used by all lsns in this test
    1740              :         pub(super) const LSN_OFFSET: Lsn = Lsn(0x08);
    1741              :         /// Number of unique keys including in the test data
    1742              :         pub(super) const KEY_COUNT: u8 = 60;
    1743              :         /// Max number of different lsns for each key
    1744              :         pub(super) const MAX_ENTRIES_PER_KEY: u8 = 20;
    1745              :         /// Possible value sizes for each key along with a probability weight
    1746              :         pub(super) const VALUE_SIZES: [(usize, u8); 3] = [(100, 2), (1024, 2), (1024 * 1024, 1)];
    1747              :         /// Probability that there will be a gap between the current key and the next one (33.3%)
    1748              :         pub(super) const KEY_GAP_CHANGES: [(bool, u8); 2] = [(true, 1), (false, 2)];
    1749              :         /// The minimum size of a key range in all the generated reads
    1750              :         pub(super) const MIN_RANGE_SIZE: i128 = 10;
    1751              :         /// The number of ranges included in each vectored read
    1752              :         pub(super) const RANGES_COUNT: u8 = 2;
    1753              :         /// The number of vectored reads performed
    1754              :         pub(super) const READS_COUNT: u8 = 100;
    1755              :         /// Soft max size of a vectored read. Will be violated if we have to read keys
    1756              :         /// with values larger than the limit
    1757              :         pub(super) const MAX_VECTORED_READ_BYTES: usize = 64 * 1024;
    1758              :     }
    1759              : 
    1760              :     struct Entry {
    1761              :         key: Key,
    1762              :         lsn: Lsn,
    1763              :         value: Vec<u8>,
    1764              :     }
    1765              : 
    1766            4 :     fn generate_entries(rng: &mut StdRng) -> Vec<Entry> {
    1767            4 :         let mut current_key = Key::MIN;
    1768            4 : 
    1769            4 :         let mut entries = Vec::new();
    1770          244 :         for _ in 0..constants::KEY_COUNT {
    1771          240 :             let count = rng.gen_range(1..constants::MAX_ENTRIES_PER_KEY);
    1772          240 :             let mut lsns_iter =
    1773         4520 :                 std::iter::successors(Some(Lsn(constants::LSN_OFFSET.0 + 0x08)), |lsn| {
    1774         4520 :                     Some(Lsn(lsn.0 + 0x08))
    1775         4520 :                 });
    1776          240 :             let mut lsns = Vec::new();
    1777         4760 :             while lsns.len() < count as usize {
    1778         4520 :                 let take = rng.gen_bool(0.5);
    1779         4520 :                 let lsn = lsns_iter.next().unwrap();
    1780         4520 :                 if take {
    1781         2224 :                     lsns.push(lsn);
    1782         2296 :                 }
    1783              :             }
    1784              : 
    1785         2464 :             for lsn in lsns {
    1786         2224 :                 let size = constants::VALUE_SIZES
    1787         6672 :                     .choose_weighted(rng, |item| item.1)
    1788         2224 :                     .unwrap()
    1789         2224 :                     .0;
    1790         2224 :                 let mut buf = vec![0; size];
    1791         2224 :                 rng.fill_bytes(&mut buf);
    1792         2224 : 
    1793         2224 :                 entries.push(Entry {
    1794         2224 :                     key: current_key,
    1795         2224 :                     lsn,
    1796         2224 :                     value: buf,
    1797         2224 :                 })
    1798              :             }
    1799              : 
    1800          240 :             let gap = constants::KEY_GAP_CHANGES
    1801          480 :                 .choose_weighted(rng, |item| item.1)
    1802          240 :                 .unwrap()
    1803          240 :                 .0;
    1804          240 :             if gap {
    1805           76 :                 current_key = current_key.add(2);
    1806          164 :             } else {
    1807          164 :                 current_key = current_key.add(1);
    1808          164 :             }
    1809              :         }
    1810              : 
    1811            4 :         entries
    1812            4 :     }
    1813              : 
    1814              :     struct EntriesMeta {
    1815              :         key_range: Range<Key>,
    1816              :         lsn_range: Range<Lsn>,
    1817              :         index: BTreeMap<(Key, Lsn), Vec<u8>>,
    1818              :     }
    1819              : 
    1820            4 :     fn get_entries_meta(entries: &[Entry]) -> EntriesMeta {
    1821         2224 :         let key_range = match entries.iter().minmax_by_key(|e| e.key) {
    1822            4 :             MinMaxResult::MinMax(min, max) => min.key..max.key.next(),
    1823            0 :             _ => panic!("More than one entry is always expected"),
    1824              :         };
    1825              : 
    1826         2224 :         let lsn_range = match entries.iter().minmax_by_key(|e| e.lsn) {
    1827            4 :             MinMaxResult::MinMax(min, max) => min.lsn..Lsn(max.lsn.0 + 1),
    1828            0 :             _ => panic!("More than one entry is always expected"),
    1829              :         };
    1830              : 
    1831            4 :         let mut index = BTreeMap::new();
    1832         2224 :         for entry in entries.iter() {
    1833         2224 :             index.insert((entry.key, entry.lsn), entry.value.clone());
    1834         2224 :         }
    1835              : 
    1836            4 :         EntriesMeta {
    1837            4 :             key_range,
    1838            4 :             lsn_range,
    1839            4 :             index,
    1840            4 :         }
    1841            4 :     }
    1842              : 
    1843          400 :     fn pick_random_keyspace(rng: &mut StdRng, key_range: &Range<Key>) -> KeySpace {
    1844          400 :         let start = key_range.start.to_i128();
    1845          400 :         let end = key_range.end.to_i128();
    1846          400 : 
    1847          400 :         let mut keyspace = KeySpace::default();
    1848              : 
    1849         1200 :         for _ in 0..constants::RANGES_COUNT {
    1850          800 :             let mut range: Option<Range<Key>> = Option::default();
    1851         2488 :             while range.is_none() || keyspace.overlaps(range.as_ref().unwrap()) {
    1852         1688 :                 let range_start = rng.gen_range(start..end);
    1853         1688 :                 let range_end_offset = range_start + constants::MIN_RANGE_SIZE;
    1854         1688 :                 if range_end_offset >= end {
    1855          200 :                     range = Some(Key::from_i128(range_start)..Key::from_i128(end));
    1856         1488 :                 } else {
    1857         1488 :                     let range_end = rng.gen_range((range_start + constants::MIN_RANGE_SIZE)..end);
    1858         1488 :                     range = Some(Key::from_i128(range_start)..Key::from_i128(range_end));
    1859         1488 :                 }
    1860              :             }
    1861          800 :             keyspace.ranges.push(range.unwrap());
    1862              :         }
    1863              : 
    1864          400 :         keyspace
    1865          400 :     }
    1866              : 
    1867              :     #[tokio::test]
    1868            4 :     async fn test_delta_layer_vectored_read_end_to_end() -> anyhow::Result<()> {
    1869            4 :         let harness = TenantHarness::create("test_delta_layer_oversized_vectored_read").await?;
    1870            4 :         let (tenant, ctx) = harness.load().await;
    1871            4 : 
    1872            4 :         let timeline_id = TimelineId::generate();
    1873            4 :         let timeline = tenant
    1874            4 :             .create_test_timeline(timeline_id, constants::LSN_OFFSET, DEFAULT_PG_VERSION, &ctx)
    1875            4 :             .await?;
    1876            4 : 
    1877            4 :         tracing::info!("Generating test data ...");
    1878            4 : 
    1879            4 :         let rng = &mut StdRng::seed_from_u64(0);
    1880            4 :         let entries = generate_entries(rng);
    1881            4 :         let entries_meta = get_entries_meta(&entries);
    1882            4 : 
    1883            4 :         tracing::info!("Done generating {} entries", entries.len());
    1884            4 : 
    1885            4 :         tracing::info!("Writing test data to delta layer ...");
    1886            4 :         let mut writer = DeltaLayerWriter::new(
    1887            4 :             harness.conf,
    1888            4 :             timeline_id,
    1889            4 :             harness.tenant_shard_id,
    1890            4 :             entries_meta.key_range.start,
    1891            4 :             entries_meta.lsn_range.clone(),
    1892            4 :             &ctx,
    1893            4 :         )
    1894            4 :         .await?;
    1895            4 : 
    1896         2228 :         for entry in entries {
    1897         2224 :             let (_, res) = writer
    1898         2224 :                 .put_value_bytes(entry.key, entry.lsn, entry.value.slice_len(), false, &ctx)
    1899         2224 :                 .await;
    1900         2224 :             res?;
    1901            4 :         }
    1902            4 : 
    1903            4 :         let (desc, path) = writer.finish(entries_meta.key_range.end, &ctx).await?;
    1904            4 :         let resident = Layer::finish_creating(harness.conf, &timeline, desc, &path)?;
    1905            4 : 
    1906            4 :         let inner = resident.get_as_delta(&ctx).await?;
    1907            4 : 
    1908            4 :         let file_size = inner.file.metadata().await?.len();
    1909            4 :         tracing::info!(
    1910            4 :             "Done writing test data to delta layer. Resulting file size is: {}",
    1911            4 :             file_size
    1912            4 :         );
    1913            4 : 
    1914          404 :         for i in 0..constants::READS_COUNT {
    1915          400 :             tracing::info!("Doing vectored read {}/{}", i + 1, constants::READS_COUNT);
    1916            4 : 
    1917          400 :             let block_reader = FileBlockReader::new(&inner.file, inner.file_id);
    1918          400 :             let index_reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
    1919          400 :                 inner.index_start_blk,
    1920          400 :                 inner.index_root_blk,
    1921          400 :                 block_reader,
    1922          400 :             );
    1923          400 : 
    1924          400 :             let planner = VectoredReadPlanner::new(constants::MAX_VECTORED_READ_BYTES);
    1925          400 :             let keyspace = pick_random_keyspace(rng, &entries_meta.key_range);
    1926          400 :             let data_end_offset = inner.index_start_blk as u64 * PAGE_SZ as u64;
    1927            4 : 
    1928          400 :             let vectored_reads = DeltaLayerInner::plan_reads(
    1929          400 :                 &keyspace,
    1930          400 :                 entries_meta.lsn_range.clone(),
    1931          400 :                 data_end_offset,
    1932          400 :                 index_reader,
    1933          400 :                 planner,
    1934          400 :                 &ctx,
    1935          400 :             )
    1936          400 :             .await?;
    1937            4 : 
    1938          400 :             let vectored_blob_reader = VectoredBlobReader::new(&inner.file);
    1939          400 :             let buf_size = DeltaLayerInner::get_min_read_buffer_size(
    1940          400 :                 &vectored_reads,
    1941          400 :                 constants::MAX_VECTORED_READ_BYTES,
    1942          400 :             );
    1943          400 :             let mut buf = Some(IoBufferMut::with_capacity(buf_size));
    1944            4 : 
    1945        39848 :             for read in vectored_reads {
    1946        39448 :                 let blobs_buf = vectored_blob_reader
    1947        39448 :                     .read_blobs(&read, buf.take().expect("Should have a buffer"), &ctx)
    1948        39448 :                     .await?;
    1949        39448 :                 let view = BufView::new_slice(&blobs_buf.buf);
    1950       114608 :                 for meta in blobs_buf.blobs.iter() {
    1951       114608 :                     let value = meta.read(&view).await?;
    1952       114608 :                     assert_eq!(
    1953       114608 :                         &value[..],
    1954       114608 :                         &entries_meta.index[&(meta.meta.key, meta.meta.lsn)]
    1955       114608 :                     );
    1956            4 :                 }
    1957            4 : 
    1958        39448 :                 buf = Some(blobs_buf.buf);
    1959            4 :             }
    1960            4 :         }
    1961            4 : 
    1962            4 :         Ok(())
    1963            4 :     }
    1964              : 
    1965              :     #[tokio::test]
    1966            4 :     async fn copy_delta_prefix_smoke() {
    1967            4 :         use bytes::Bytes;
    1968            4 :         use pageserver_api::record::NeonWalRecord;
    1969            4 : 
    1970            4 :         let h = crate::tenant::harness::TenantHarness::create("truncate_delta_smoke")
    1971            4 :             .await
    1972            4 :             .unwrap();
    1973            4 :         let (tenant, ctx) = h.load().await;
    1974            4 :         let ctx = &ctx;
    1975            4 :         let timeline = tenant
    1976            4 :             .create_test_timeline(TimelineId::generate(), Lsn(0x10), 14, ctx)
    1977            4 :             .await
    1978            4 :             .unwrap();
    1979            4 : 
    1980            4 :         let initdb_layer = timeline
    1981            4 :             .layers
    1982            4 :             .read()
    1983            4 :             .await
    1984            4 :             .likely_resident_layers()
    1985            4 :             .next()
    1986            4 :             .cloned()
    1987            4 :             .unwrap();
    1988            4 : 
    1989            4 :         {
    1990            4 :             let mut writer = timeline.writer().await;
    1991            4 : 
    1992            4 :             let data = [
    1993            4 :                 (0x20, 12, Value::Image(Bytes::from_static(b"foobar"))),
    1994            4 :                 (
    1995            4 :                     0x30,
    1996            4 :                     12,
    1997            4 :                     Value::WalRecord(NeonWalRecord::Postgres {
    1998            4 :                         will_init: false,
    1999            4 :                         rec: Bytes::from_static(b"1"),
    2000            4 :                     }),
    2001            4 :                 ),
    2002            4 :                 (
    2003            4 :                     0x40,
    2004            4 :                     12,
    2005            4 :                     Value::WalRecord(NeonWalRecord::Postgres {
    2006            4 :                         will_init: true,
    2007            4 :                         rec: Bytes::from_static(b"2"),
    2008            4 :                     }),
    2009            4 :                 ),
    2010            4 :                 // build an oversized value so we cannot extend and existing read over
    2011            4 :                 // this
    2012            4 :                 (
    2013            4 :                     0x50,
    2014            4 :                     12,
    2015            4 :                     Value::WalRecord(NeonWalRecord::Postgres {
    2016            4 :                         will_init: true,
    2017            4 :                         rec: {
    2018            4 :                             let mut buf =
    2019            4 :                                 vec![0u8; tenant.conf.max_vectored_read_bytes.0.get() + 1024];
    2020            4 :                             buf.iter_mut()
    2021            4 :                                 .enumerate()
    2022       536576 :                                 .for_each(|(i, slot)| *slot = (i % 256) as u8);
    2023            4 :                             Bytes::from(buf)
    2024            4 :                         },
    2025            4 :                     }),
    2026            4 :                 ),
    2027            4 :                 // because the oversized read cannot be extended further, we are sure to exercise the
    2028            4 :                 // builder created on the last round with this:
    2029            4 :                 (
    2030            4 :                     0x60,
    2031            4 :                     12,
    2032            4 :                     Value::WalRecord(NeonWalRecord::Postgres {
    2033            4 :                         will_init: true,
    2034            4 :                         rec: Bytes::from_static(b"3"),
    2035            4 :                     }),
    2036            4 :                 ),
    2037            4 :                 (
    2038            4 :                     0x60,
    2039            4 :                     9,
    2040            4 :                     Value::Image(Bytes::from_static(b"something for a different key")),
    2041            4 :                 ),
    2042            4 :             ];
    2043            4 : 
    2044            4 :             let mut last_lsn = None;
    2045            4 : 
    2046           28 :             for (lsn, key, value) in data {
    2047           24 :                 let key = Key::from_i128(key);
    2048           24 :                 writer.put(key, Lsn(lsn), &value, ctx).await.unwrap();
    2049           24 :                 last_lsn = Some(lsn);
    2050            4 :             }
    2051            4 : 
    2052            4 :             writer.finish_write(Lsn(last_lsn.unwrap()));
    2053            4 :         }
    2054            4 :         timeline.freeze_and_flush().await.unwrap();
    2055            4 : 
    2056            4 :         let new_layer = timeline
    2057            4 :             .layers
    2058            4 :             .read()
    2059            4 :             .await
    2060            4 :             .likely_resident_layers()
    2061            5 :             .find(|&x| x != &initdb_layer)
    2062            4 :             .cloned()
    2063            4 :             .unwrap();
    2064            4 : 
    2065            4 :         // create a copy for the timeline, so we don't overwrite the file
    2066            4 :         let branch = tenant
    2067            4 :             .branch_timeline_test(&timeline, TimelineId::generate(), None, ctx)
    2068            4 :             .await
    2069            4 :             .unwrap();
    2070            4 : 
    2071            4 :         assert_eq!(branch.get_ancestor_lsn(), Lsn(0x60));
    2072            4 : 
    2073            4 :         // truncating at 0x61 gives us a full copy, otherwise just go backwards until there's just
    2074            4 :         // a single key
    2075            4 : 
    2076           24 :         for truncate_at in [0x61, 0x51, 0x41, 0x31, 0x21] {
    2077           20 :             let truncate_at = Lsn(truncate_at);
    2078            4 : 
    2079           20 :             let mut writer = DeltaLayerWriter::new(
    2080           20 :                 tenant.conf,
    2081           20 :                 branch.timeline_id,
    2082           20 :                 tenant.tenant_shard_id,
    2083           20 :                 Key::MIN,
    2084           20 :                 Lsn(0x11)..truncate_at,
    2085           20 :                 ctx,
    2086           20 :             )
    2087           20 :             .await
    2088           20 :             .unwrap();
    2089            4 : 
    2090           20 :             let new_layer = new_layer.download_and_keep_resident().await.unwrap();
    2091           20 : 
    2092           20 :             new_layer
    2093           20 :                 .copy_delta_prefix(&mut writer, truncate_at, ctx)
    2094           20 :                 .await
    2095           20 :                 .unwrap();
    2096            4 : 
    2097           20 :             let (desc, path) = writer.finish(Key::MAX, ctx).await.unwrap();
    2098           20 :             let copied_layer = Layer::finish_creating(tenant.conf, &branch, desc, &path).unwrap();
    2099           20 : 
    2100           20 :             copied_layer.get_as_delta(ctx).await.unwrap();
    2101           20 : 
    2102           20 :             assert_keys_and_values_eq(
    2103           20 :                 new_layer.get_as_delta(ctx).await.unwrap(),
    2104           20 :                 copied_layer.get_as_delta(ctx).await.unwrap(),
    2105           20 :                 truncate_at,
    2106           20 :                 ctx,
    2107           20 :             )
    2108           20 :             .await;
    2109            4 :         }
    2110            4 :     }
    2111              : 
    2112           20 :     async fn assert_keys_and_values_eq(
    2113           20 :         source: &DeltaLayerInner,
    2114           20 :         truncated: &DeltaLayerInner,
    2115           20 :         truncated_at: Lsn,
    2116           20 :         ctx: &RequestContext,
    2117           20 :     ) {
    2118              :         use futures::future::ready;
    2119              :         use futures::stream::TryStreamExt;
    2120              : 
    2121           20 :         let start_key = [0u8; DELTA_KEY_SIZE];
    2122           20 : 
    2123           20 :         let source_reader = FileBlockReader::new(&source.file, source.file_id);
    2124           20 :         let source_tree = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
    2125           20 :             source.index_start_blk,
    2126           20 :             source.index_root_blk,
    2127           20 :             &source_reader,
    2128           20 :         );
    2129           20 :         let source_stream = source.stream_index_forwards(source_tree, &start_key, ctx);
    2130          120 :         let source_stream = source_stream.filter(|res| match res {
    2131          120 :             Ok((_, lsn, _)) => ready(lsn < &truncated_at),
    2132            0 :             _ => ready(true),
    2133          120 :         });
    2134           20 :         let mut source_stream = std::pin::pin!(source_stream);
    2135           20 : 
    2136           20 :         let truncated_reader = FileBlockReader::new(&truncated.file, truncated.file_id);
    2137           20 :         let truncated_tree = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
    2138           20 :             truncated.index_start_blk,
    2139           20 :             truncated.index_root_blk,
    2140           20 :             &truncated_reader,
    2141           20 :         );
    2142           20 :         let truncated_stream = truncated.stream_index_forwards(truncated_tree, &start_key, ctx);
    2143           20 :         let mut truncated_stream = std::pin::pin!(truncated_stream);
    2144           20 : 
    2145           20 :         let mut scratch_left = Vec::new();
    2146           20 :         let mut scratch_right = Vec::new();
    2147              : 
    2148              :         loop {
    2149           84 :             let (src, truncated) = (source_stream.try_next(), truncated_stream.try_next());
    2150           84 :             let (src, truncated) = tokio::try_join!(src, truncated).unwrap();
    2151           84 : 
    2152           84 :             if src.is_none() {
    2153           20 :                 assert!(truncated.is_none());
    2154           20 :                 break;
    2155           64 :             }
    2156           64 : 
    2157           64 :             let (src, truncated) = (src.unwrap(), truncated.unwrap());
    2158           64 : 
    2159           64 :             // because we've filtered the source with Lsn, we should always have the same keys from both.
    2160           64 :             assert_eq!(src.0, truncated.0);
    2161           64 :             assert_eq!(src.1, truncated.1);
    2162              : 
    2163              :             // if this is needed for something else, just drop this assert.
    2164           64 :             assert!(
    2165           64 :                 src.2.pos() >= truncated.2.pos(),
    2166            0 :                 "value position should not go backwards {} vs. {}",
    2167            0 :                 src.2.pos(),
    2168            0 :                 truncated.2.pos()
    2169              :             );
    2170              : 
    2171           64 :             scratch_left.clear();
    2172           64 :             let src_cursor = source_reader.block_cursor();
    2173           64 :             let left = src_cursor.read_blob_into_buf(src.2.pos(), &mut scratch_left, ctx);
    2174           64 :             scratch_right.clear();
    2175           64 :             let trunc_cursor = truncated_reader.block_cursor();
    2176           64 :             let right = trunc_cursor.read_blob_into_buf(truncated.2.pos(), &mut scratch_right, ctx);
    2177           64 : 
    2178           64 :             tokio::try_join!(left, right).unwrap();
    2179           64 : 
    2180           64 :             assert_eq!(utils::Hex(&scratch_left), utils::Hex(&scratch_right));
    2181              :         }
    2182           20 :     }
    2183              : 
    2184        36448 :     pub(crate) fn sort_delta(
    2185        36448 :         (k1, l1, _): &(Key, Lsn, Value),
    2186        36448 :         (k2, l2, _): &(Key, Lsn, Value),
    2187        36448 :     ) -> std::cmp::Ordering {
    2188        36448 :         (k1, l1).cmp(&(k2, l2))
    2189        36448 :     }
    2190              : 
    2191              :     #[cfg(feature = "testing")]
    2192          188 :     pub(crate) fn sort_delta_value(
    2193          188 :         (k1, l1, v1): &(Key, Lsn, Value),
    2194          188 :         (k2, l2, v2): &(Key, Lsn, Value),
    2195          188 :     ) -> std::cmp::Ordering {
    2196          188 :         let order_1 = if v1.is_image() { 0 } else { 1 };
    2197          188 :         let order_2 = if v2.is_image() { 0 } else { 1 };
    2198          188 :         (k1, l1, order_1).cmp(&(k2, l2, order_2))
    2199          188 :     }
    2200              : 
    2201           44 :     pub(crate) async fn produce_delta_layer(
    2202           44 :         tenant: &Tenant,
    2203           44 :         tline: &Arc<Timeline>,
    2204           44 :         mut deltas: Vec<(Key, Lsn, Value)>,
    2205           44 :         ctx: &RequestContext,
    2206           44 :     ) -> anyhow::Result<ResidentLayer> {
    2207           44 :         deltas.sort_by(sort_delta);
    2208           44 :         let (key_start, _, _) = deltas.first().unwrap();
    2209           44 :         let (key_max, _, _) = deltas.last().unwrap();
    2210        16480 :         let lsn_min = deltas.iter().map(|(_, lsn, _)| lsn).min().unwrap();
    2211        16480 :         let lsn_max = deltas.iter().map(|(_, lsn, _)| lsn).max().unwrap();
    2212           44 :         let lsn_end = Lsn(lsn_max.0 + 1);
    2213           44 :         let mut writer = DeltaLayerWriter::new(
    2214           44 :             tenant.conf,
    2215           44 :             tline.timeline_id,
    2216           44 :             tenant.tenant_shard_id,
    2217           44 :             *key_start,
    2218           44 :             (*lsn_min)..lsn_end,
    2219           44 :             ctx,
    2220           44 :         )
    2221           44 :         .await?;
    2222           44 :         let key_end = key_max.next();
    2223              : 
    2224        16524 :         for (key, lsn, value) in deltas {
    2225        16480 :             writer.put_value(key, lsn, value, ctx).await?;
    2226              :         }
    2227              : 
    2228           44 :         let (desc, path) = writer.finish(key_end, ctx).await?;
    2229           44 :         let delta_layer = Layer::finish_creating(tenant.conf, tline, desc, &path)?;
    2230              : 
    2231           44 :         Ok::<_, anyhow::Error>(delta_layer)
    2232           44 :     }
    2233              : 
    2234           56 :     async fn assert_delta_iter_equal(
    2235           56 :         delta_iter: &mut DeltaLayerIterator<'_>,
    2236           56 :         expect: &[(Key, Lsn, Value)],
    2237           56 :     ) {
    2238           56 :         let mut expect_iter = expect.iter();
    2239              :         loop {
    2240        56056 :             let o1 = delta_iter.next().await.unwrap();
    2241        56056 :             let o2 = expect_iter.next();
    2242        56056 :             assert_eq!(o1.is_some(), o2.is_some());
    2243        56056 :             if o1.is_none() && o2.is_none() {
    2244           56 :                 break;
    2245        56000 :             }
    2246        56000 :             let (k1, l1, v1) = o1.unwrap();
    2247        56000 :             let (k2, l2, v2) = o2.unwrap();
    2248        56000 :             assert_eq!(&k1, k2);
    2249        56000 :             assert_eq!(l1, *l2);
    2250        56000 :             assert_eq!(&v1, v2);
    2251              :         }
    2252           56 :     }
    2253              : 
    2254              :     #[tokio::test]
    2255            4 :     async fn delta_layer_iterator() {
    2256            4 :         let harness = TenantHarness::create("delta_layer_iterator").await.unwrap();
    2257            4 :         let (tenant, ctx) = harness.load().await;
    2258            4 : 
    2259            4 :         let tline = tenant
    2260            4 :             .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
    2261            4 :             .await
    2262            4 :             .unwrap();
    2263            4 : 
    2264         4000 :         fn get_key(id: u32) -> Key {
    2265         4000 :             let mut key = Key::from_hex("000000000033333333444444445500000000").unwrap();
    2266         4000 :             key.field6 = id;
    2267         4000 :             key
    2268         4000 :         }
    2269            4 :         const N: usize = 1000;
    2270            4 :         let test_deltas = (0..N)
    2271         4000 :             .map(|idx| {
    2272         4000 :                 (
    2273         4000 :                     get_key(idx as u32 / 10),
    2274         4000 :                     Lsn(0x10 * ((idx as u64) % 10 + 1)),
    2275         4000 :                     Value::Image(Bytes::from(format!("img{idx:05}"))),
    2276         4000 :                 )
    2277         4000 :             })
    2278            4 :             .collect_vec();
    2279            4 :         let resident_layer = produce_delta_layer(&tenant, &tline, test_deltas.clone(), &ctx)
    2280            4 :             .await
    2281            4 :             .unwrap();
    2282            4 :         let delta_layer = resident_layer.get_as_delta(&ctx).await.unwrap();
    2283           12 :         for max_read_size in [1, 1024] {
    2284           64 :             for batch_size in [1, 2, 4, 8, 3, 7, 13] {
    2285           56 :                 println!("running with batch_size={batch_size} max_read_size={max_read_size}");
    2286           56 :                 // Test if the batch size is correctly determined
    2287           56 :                 let mut iter = delta_layer.iter(&ctx);
    2288           56 :                 iter.planner = StreamingVectoredReadPlanner::new(max_read_size, batch_size);
    2289           56 :                 let mut num_items = 0;
    2290          224 :                 for _ in 0..3 {
    2291          168 :                     iter.next_batch().await.unwrap();
    2292          168 :                     num_items += iter.key_values_batch.len();
    2293          168 :                     if max_read_size == 1 {
    2294            4 :                         // every key should be a batch b/c the value is larger than max_read_size
    2295           84 :                         assert_eq!(iter.key_values_batch.len(), 1);
    2296            4 :                     } else {
    2297           84 :                         assert!(iter.key_values_batch.len() <= batch_size);
    2298            4 :                     }
    2299          168 :                     if num_items >= N {
    2300            4 :                         break;
    2301          168 :                     }
    2302          168 :                     iter.key_values_batch.clear();
    2303            4 :                 }
    2304            4 :                 // Test if the result is correct
    2305           56 :                 let mut iter = delta_layer.iter(&ctx);
    2306           56 :                 iter.planner = StreamingVectoredReadPlanner::new(max_read_size, batch_size);
    2307           56 :                 assert_delta_iter_equal(&mut iter, &test_deltas).await;
    2308            4 :             }
    2309            4 :         }
    2310            4 :     }
    2311              : }
        

Generated by: LCOV version 2.1-beta