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

Generated by: LCOV version 2.1-beta