LCOV - code coverage report
Current view: top level - pageserver/src/tenant/timeline - compaction.rs (source / functions) Coverage Total Hit
Test: f8d8f5b90fa487a9e82c42da223f012f5d4fece7.info Lines: 60.7 % 1683 1021
Test Date: 2024-09-19 20:36:02 Functions: 40.0 % 130 52

            Line data    Source code
       1              : //! New compaction implementation. The algorithm itself is implemented in the
       2              : //! compaction crate. This file implements the callbacks and structs that allow
       3              : //! the algorithm to drive the process.
       4              : //!
       5              : //! The old legacy algorithm is implemented directly in `timeline.rs`.
       6              : 
       7              : use std::collections::{BinaryHeap, HashSet};
       8              : use std::ops::{Deref, Range};
       9              : use std::sync::Arc;
      10              : 
      11              : use super::layer_manager::LayerManager;
      12              : use super::{
      13              :     CompactFlags, CreateImageLayersError, DurationRecorder, ImageLayerCreationMode,
      14              :     RecordedDuration, Timeline,
      15              : };
      16              : 
      17              : use anyhow::{anyhow, bail, Context};
      18              : use bytes::Bytes;
      19              : use enumset::EnumSet;
      20              : use fail::fail_point;
      21              : use itertools::Itertools;
      22              : use pageserver_api::key::KEY_SIZE;
      23              : use pageserver_api::keyspace::ShardedRange;
      24              : use pageserver_api::shard::{ShardCount, ShardIdentity, TenantShardId};
      25              : use serde::Serialize;
      26              : use tokio_util::sync::CancellationToken;
      27              : use tracing::{debug, info, info_span, trace, warn, Instrument};
      28              : use utils::id::TimelineId;
      29              : 
      30              : use crate::context::{AccessStatsBehavior, RequestContext, RequestContextBuilder};
      31              : use crate::page_cache;
      32              : use crate::tenant::checks::check_valid_layermap;
      33              : use crate::tenant::remote_timeline_client::WaitCompletionError;
      34              : use crate::tenant::storage_layer::merge_iterator::MergeIterator;
      35              : use crate::tenant::storage_layer::split_writer::{
      36              :     SplitDeltaLayerWriter, SplitImageLayerWriter, SplitWriterResult,
      37              : };
      38              : use crate::tenant::storage_layer::{
      39              :     AsLayerDesc, PersistentLayerDesc, PersistentLayerKey, ValueReconstructState,
      40              : };
      41              : use crate::tenant::timeline::ImageLayerCreationOutcome;
      42              : use crate::tenant::timeline::{drop_rlock, DeltaLayerWriter, ImageLayerWriter};
      43              : use crate::tenant::timeline::{Layer, ResidentLayer};
      44              : use crate::tenant::DeltaLayer;
      45              : use crate::virtual_file::{MaybeFatalIo, VirtualFile};
      46              : use pageserver_api::config::tenant_conf_defaults::{
      47              :     DEFAULT_CHECKPOINT_DISTANCE, DEFAULT_COMPACTION_THRESHOLD,
      48              : };
      49              : 
      50              : use crate::keyspace::KeySpace;
      51              : use crate::repository::{Key, Value};
      52              : use crate::walrecord::NeonWalRecord;
      53              : 
      54              : use utils::lsn::Lsn;
      55              : 
      56              : use pageserver_compaction::helpers::overlaps_with;
      57              : use pageserver_compaction::interface::*;
      58              : 
      59              : use super::CompactionError;
      60              : 
      61              : /// Maximum number of deltas before generating an image layer in bottom-most compaction.
      62              : const COMPACTION_DELTA_THRESHOLD: usize = 5;
      63              : 
      64              : /// The result of bottom-most compaction for a single key at each LSN.
      65              : #[derive(Debug)]
      66              : #[cfg_attr(test, derive(PartialEq))]
      67              : pub struct KeyLogAtLsn(pub Vec<(Lsn, Value)>);
      68              : 
      69              : /// The result of bottom-most compaction.
      70              : #[derive(Debug)]
      71              : #[cfg_attr(test, derive(PartialEq))]
      72              : pub(crate) struct KeyHistoryRetention {
      73              :     /// Stores logs to reconstruct the value at the given LSN, that is to say, logs <= LSN or image == LSN.
      74              :     pub(crate) below_horizon: Vec<(Lsn, KeyLogAtLsn)>,
      75              :     /// Stores logs to reconstruct the value at any LSN above the horizon, that is to say, log > LSN.
      76              :     pub(crate) above_horizon: KeyLogAtLsn,
      77              : }
      78              : 
      79              : impl KeyHistoryRetention {
      80              :     /// Hack: skip delta layer if we need to produce a layer of a same key-lsn.
      81              :     ///
      82              :     /// This can happen if we have removed some deltas in "the middle" of some existing layer's key-lsn-range.
      83              :     /// For example, consider the case where a single delta with range [0x10,0x50) exists.
      84              :     /// And we have branches at LSN 0x10, 0x20, 0x30.
      85              :     /// Then we delete branch @ 0x20.
      86              :     /// Bottom-most compaction may now delete the delta [0x20,0x30).
      87              :     /// And that wouldnt' change the shape of the layer.
      88              :     ///
      89              :     /// Note that bottom-most-gc-compaction never _adds_ new data in that case, only removes.
      90              :     ///
      91              :     /// `discard_key` will only be called when the writer reaches its target (instead of for every key), so it's fine to grab a lock inside.
      92          114 :     async fn discard_key(key: &PersistentLayerKey, tline: &Arc<Timeline>, dry_run: bool) -> bool {
      93          114 :         if dry_run {
      94            0 :             return true;
      95          114 :         }
      96          114 :         let guard = tline.layers.read().await;
      97          114 :         if !guard.contains_key(key) {
      98           66 :             return false;
      99           48 :         }
     100           48 :         let layer_generation = guard.get_from_key(key).metadata().generation;
     101           48 :         drop(guard);
     102           48 :         if layer_generation == tline.generation {
     103           48 :             info!(
     104              :                 key=%key,
     105              :                 ?layer_generation,
     106            0 :                 "discard layer due to duplicated layer key in the same generation",
     107              :             );
     108           48 :             true
     109              :         } else {
     110            0 :             false
     111              :         }
     112          114 :     }
     113              : 
     114              :     /// Pipe a history of a single key to the writers.
     115              :     ///
     116              :     /// If `image_writer` is none, the images will be placed into the delta layers.
     117              :     /// The delta writer will contain all images and deltas (below and above the horizon) except the bottom-most images.
     118              :     #[allow(clippy::too_many_arguments)]
     119         1266 :     async fn pipe_to(
     120         1266 :         self,
     121         1266 :         key: Key,
     122         1266 :         tline: &Arc<Timeline>,
     123         1266 :         delta_writer: &mut SplitDeltaLayerWriter,
     124         1266 :         mut image_writer: Option<&mut SplitImageLayerWriter>,
     125         1266 :         stat: &mut CompactionStatistics,
     126         1266 :         dry_run: bool,
     127         1266 :         ctx: &RequestContext,
     128         1266 :     ) -> anyhow::Result<()> {
     129         1266 :         let mut first_batch = true;
     130         1266 :         let discard = |key: &PersistentLayerKey| {
     131            0 :             let key = key.clone();
     132            0 :             async move { Self::discard_key(&key, tline, dry_run).await }
     133            0 :         };
     134         4206 :         for (cutoff_lsn, KeyLogAtLsn(logs)) in self.below_horizon {
     135         2940 :             if first_batch {
     136         1266 :                 if logs.len() == 1 && logs[0].1.is_image() {
     137         1224 :                     let Value::Image(img) = &logs[0].1 else {
     138            0 :                         unreachable!()
     139              :                     };
     140         1224 :                     stat.produce_image_key(img);
     141         1224 :                     if let Some(image_writer) = image_writer.as_mut() {
     142         1224 :                         image_writer
     143         1224 :                             .put_image_with_discard_fn(key, img.clone(), tline, ctx, discard)
     144         1242 :                             .await?;
     145              :                     } else {
     146            0 :                         delta_writer
     147            0 :                             .put_value_with_discard_fn(
     148            0 :                                 key,
     149            0 :                                 cutoff_lsn,
     150            0 :                                 Value::Image(img.clone()),
     151            0 :                                 tline,
     152            0 :                                 ctx,
     153            0 :                                 discard,
     154            0 :                             )
     155            0 :                             .await?;
     156              :                     }
     157              :                 } else {
     158           84 :                     for (lsn, val) in logs {
     159           42 :                         stat.produce_key(&val);
     160           42 :                         delta_writer
     161           42 :                             .put_value_with_discard_fn(key, lsn, val, tline, ctx, discard)
     162            3 :                             .await?;
     163              :                     }
     164              :                 }
     165         1266 :                 first_batch = false;
     166              :             } else {
     167         1920 :                 for (lsn, val) in logs {
     168          246 :                     stat.produce_key(&val);
     169          246 :                     delta_writer
     170          246 :                         .put_value_with_discard_fn(key, lsn, val, tline, ctx, discard)
     171           24 :                         .await?;
     172              :                 }
     173              :             }
     174              :         }
     175         1266 :         let KeyLogAtLsn(above_horizon_logs) = self.above_horizon;
     176         1362 :         for (lsn, val) in above_horizon_logs {
     177           96 :             stat.produce_key(&val);
     178           96 :             delta_writer
     179           96 :                 .put_value_with_discard_fn(key, lsn, val, tline, ctx, discard)
     180            6 :                 .await?;
     181              :         }
     182         1266 :         Ok(())
     183         1266 :     }
     184              : }
     185              : 
     186              : #[derive(Debug, Serialize, Default)]
     187              : struct CompactionStatisticsNumSize {
     188              :     num: u64,
     189              :     size: u64,
     190              : }
     191              : 
     192              : #[derive(Debug, Serialize, Default)]
     193              : pub struct CompactionStatistics {
     194              :     delta_layer_visited: CompactionStatisticsNumSize,
     195              :     image_layer_visited: CompactionStatisticsNumSize,
     196              :     delta_layer_produced: CompactionStatisticsNumSize,
     197              :     image_layer_produced: CompactionStatisticsNumSize,
     198              :     num_delta_layer_discarded: usize,
     199              :     num_image_layer_discarded: usize,
     200              :     num_unique_keys_visited: usize,
     201              :     wal_keys_visited: CompactionStatisticsNumSize,
     202              :     image_keys_visited: CompactionStatisticsNumSize,
     203              :     wal_produced: CompactionStatisticsNumSize,
     204              :     image_produced: CompactionStatisticsNumSize,
     205              : }
     206              : 
     207              : impl CompactionStatistics {
     208         2058 :     fn estimated_size_of_value(val: &Value) -> usize {
     209          798 :         match val {
     210         1260 :             Value::Image(img) => img.len(),
     211            0 :             Value::WalRecord(NeonWalRecord::Postgres { rec, .. }) => rec.len(),
     212          798 :             _ => std::mem::size_of::<NeonWalRecord>(),
     213              :         }
     214         2058 :     }
     215         3288 :     fn estimated_size_of_key() -> usize {
     216         3288 :         KEY_SIZE // TODO: distinguish image layer and delta layer (count LSN in delta layer)
     217         3288 :     }
     218          138 :     fn visit_delta_layer(&mut self, size: u64) {
     219          138 :         self.delta_layer_visited.num += 1;
     220          138 :         self.delta_layer_visited.size += size;
     221          138 :     }
     222          108 :     fn visit_image_layer(&mut self, size: u64) {
     223          108 :         self.image_layer_visited.num += 1;
     224          108 :         self.image_layer_visited.size += size;
     225          108 :     }
     226         1266 :     fn on_unique_key_visited(&mut self) {
     227         1266 :         self.num_unique_keys_visited += 1;
     228         1266 :     }
     229          420 :     fn visit_wal_key(&mut self, val: &Value) {
     230          420 :         self.wal_keys_visited.num += 1;
     231          420 :         self.wal_keys_visited.size +=
     232          420 :             Self::estimated_size_of_value(val) as u64 + Self::estimated_size_of_key() as u64;
     233          420 :     }
     234         1260 :     fn visit_image_key(&mut self, val: &Value) {
     235         1260 :         self.image_keys_visited.num += 1;
     236         1260 :         self.image_keys_visited.size +=
     237         1260 :             Self::estimated_size_of_value(val) as u64 + Self::estimated_size_of_key() as u64;
     238         1260 :     }
     239          384 :     fn produce_key(&mut self, val: &Value) {
     240          384 :         match val {
     241            6 :             Value::Image(img) => self.produce_image_key(img),
     242          378 :             Value::WalRecord(_) => self.produce_wal_key(val),
     243              :         }
     244          384 :     }
     245          378 :     fn produce_wal_key(&mut self, val: &Value) {
     246          378 :         self.wal_produced.num += 1;
     247          378 :         self.wal_produced.size +=
     248          378 :             Self::estimated_size_of_value(val) as u64 + Self::estimated_size_of_key() as u64;
     249          378 :     }
     250         1230 :     fn produce_image_key(&mut self, val: &Bytes) {
     251         1230 :         self.image_produced.num += 1;
     252         1230 :         self.image_produced.size += val.len() as u64 + Self::estimated_size_of_key() as u64;
     253         1230 :     }
     254           24 :     fn discard_delta_layer(&mut self) {
     255           24 :         self.num_delta_layer_discarded += 1;
     256           24 :     }
     257           24 :     fn discard_image_layer(&mut self) {
     258           24 :         self.num_image_layer_discarded += 1;
     259           24 :     }
     260           30 :     fn produce_delta_layer(&mut self, size: u64) {
     261           30 :         self.delta_layer_produced.num += 1;
     262           30 :         self.delta_layer_produced.size += size;
     263           30 :     }
     264           36 :     fn produce_image_layer(&mut self, size: u64) {
     265           36 :         self.image_layer_produced.num += 1;
     266           36 :         self.image_layer_produced.size += size;
     267           36 :     }
     268              : }
     269              : 
     270              : impl Timeline {
     271              :     /// TODO: cancellation
     272              :     ///
     273              :     /// Returns whether the compaction has pending tasks.
     274         1092 :     pub(crate) async fn compact_legacy(
     275         1092 :         self: &Arc<Self>,
     276         1092 :         cancel: &CancellationToken,
     277         1092 :         flags: EnumSet<CompactFlags>,
     278         1092 :         ctx: &RequestContext,
     279         1092 :     ) -> Result<bool, CompactionError> {
     280         1092 :         if flags.contains(CompactFlags::EnhancedGcBottomMostCompaction) {
     281            0 :             self.compact_with_gc(cancel, flags, ctx)
     282            0 :                 .await
     283            0 :                 .map_err(CompactionError::Other)?;
     284            0 :             return Ok(false);
     285         1092 :         }
     286         1092 : 
     287         1092 :         if flags.contains(CompactFlags::DryRun) {
     288            0 :             return Err(CompactionError::Other(anyhow!(
     289            0 :                 "dry-run mode is not supported for legacy compaction for now"
     290            0 :             )));
     291         1092 :         }
     292         1092 : 
     293         1092 :         // High level strategy for compaction / image creation:
     294         1092 :         //
     295         1092 :         // 1. First, calculate the desired "partitioning" of the
     296         1092 :         // currently in-use key space. The goal is to partition the
     297         1092 :         // key space into roughly fixed-size chunks, but also take into
     298         1092 :         // account any existing image layers, and try to align the
     299         1092 :         // chunk boundaries with the existing image layers to avoid
     300         1092 :         // too much churn. Also try to align chunk boundaries with
     301         1092 :         // relation boundaries.  In principle, we don't know about
     302         1092 :         // relation boundaries here, we just deal with key-value
     303         1092 :         // pairs, and the code in pgdatadir_mapping.rs knows how to
     304         1092 :         // map relations into key-value pairs. But in practice we know
     305         1092 :         // that 'field6' is the block number, and the fields 1-5
     306         1092 :         // identify a relation. This is just an optimization,
     307         1092 :         // though.
     308         1092 :         //
     309         1092 :         // 2. Once we know the partitioning, for each partition,
     310         1092 :         // decide if it's time to create a new image layer. The
     311         1092 :         // criteria is: there has been too much "churn" since the last
     312         1092 :         // image layer? The "churn" is fuzzy concept, it's a
     313         1092 :         // combination of too many delta files, or too much WAL in
     314         1092 :         // total in the delta file. Or perhaps: if creating an image
     315         1092 :         // file would allow to delete some older files.
     316         1092 :         //
     317         1092 :         // 3. After that, we compact all level0 delta files if there
     318         1092 :         // are too many of them.  While compacting, we also garbage
     319         1092 :         // collect any page versions that are no longer needed because
     320         1092 :         // of the new image layers we created in step 2.
     321         1092 :         //
     322         1092 :         // TODO: This high level strategy hasn't been implemented yet.
     323         1092 :         // Below are functions compact_level0() and create_image_layers()
     324         1092 :         // but they are a bit ad hoc and don't quite work like it's explained
     325         1092 :         // above. Rewrite it.
     326         1092 : 
     327         1092 :         // Is the timeline being deleted?
     328         1092 :         if self.is_stopping() {
     329            0 :             trace!("Dropping out of compaction on timeline shutdown");
     330            0 :             return Err(CompactionError::ShuttingDown);
     331         1092 :         }
     332         1092 : 
     333         1092 :         let target_file_size = self.get_checkpoint_distance();
     334              : 
     335              :         // Define partitioning schema if needed
     336              : 
     337              :         // FIXME: the match should only cover repartitioning, not the next steps
     338         1092 :         let (partition_count, has_pending_tasks) = match self
     339         1092 :             .repartition(
     340         1092 :                 self.get_last_record_lsn(),
     341         1092 :                 self.get_compaction_target_size(),
     342         1092 :                 flags,
     343         1092 :                 ctx,
     344         1092 :             )
     345        47978 :             .await
     346              :         {
     347         1092 :             Ok(((dense_partitioning, sparse_partitioning), lsn)) => {
     348         1092 :                 // Disables access_stats updates, so that the files we read remain candidates for eviction after we're done with them
     349         1092 :                 let image_ctx = RequestContextBuilder::extend(ctx)
     350         1092 :                     .access_stats_behavior(AccessStatsBehavior::Skip)
     351         1092 :                     .build();
     352         1092 : 
     353         1092 :                 // 2. Compact
     354         1092 :                 let timer = self.metrics.compact_time_histo.start_timer();
     355        29652 :                 let fully_compacted = self.compact_level0(target_file_size, ctx).await?;
     356         1092 :                 timer.stop_and_record();
     357         1092 : 
     358         1092 :                 let mut partitioning = dense_partitioning;
     359         1092 :                 partitioning
     360         1092 :                     .parts
     361         1092 :                     .extend(sparse_partitioning.into_dense().parts);
     362         1092 : 
     363         1092 :                 // 3. Create new image layers for partitions that have been modified
     364         1092 :                 // "enough". Skip image layer creation if L0 compaction cannot keep up.
     365         1092 :                 if fully_compacted {
     366         1092 :                     let image_layers = self
     367         1092 :                         .create_image_layers(
     368         1092 :                             &partitioning,
     369         1092 :                             lsn,
     370         1092 :                             if flags.contains(CompactFlags::ForceImageLayerCreation) {
     371           42 :                                 ImageLayerCreationMode::Force
     372              :                             } else {
     373         1050 :                                 ImageLayerCreationMode::Try
     374              :                             },
     375         1092 :                             &image_ctx,
     376              :                         )
     377        40906 :                         .await?;
     378              : 
     379         1092 :                     self.upload_new_image_layers(image_layers)?;
     380              :                 } else {
     381            0 :                     info!("skipping image layer generation due to L0 compaction did not include all layers.");
     382              :                 }
     383         1092 :                 (partitioning.parts.len(), !fully_compacted)
     384              :             }
     385            0 :             Err(err) => {
     386            0 :                 // no partitioning? This is normal, if the timeline was just created
     387            0 :                 // as an empty timeline. Also in unit tests, when we use the timeline
     388            0 :                 // as a simple key-value store, ignoring the datadir layout. Log the
     389            0 :                 // error but continue.
     390            0 :                 //
     391            0 :                 // Suppress error when it's due to cancellation
     392            0 :                 if !self.cancel.is_cancelled() {
     393            0 :                     tracing::error!("could not compact, repartitioning keyspace failed: {err:?}");
     394            0 :                 }
     395            0 :                 (1, false)
     396              :             }
     397              :         };
     398              : 
     399         1092 :         if self.shard_identity.count >= ShardCount::new(2) {
     400              :             // Limit the number of layer rewrites to the number of partitions: this means its
     401              :             // runtime should be comparable to a full round of image layer creations, rather than
     402              :             // being potentially much longer.
     403            0 :             let rewrite_max = partition_count;
     404            0 : 
     405            0 :             self.compact_shard_ancestors(rewrite_max, ctx).await?;
     406         1092 :         }
     407              : 
     408         1092 :         Ok(has_pending_tasks)
     409         1092 :     }
     410              : 
     411              :     /// Check for layers that are elegible to be rewritten:
     412              :     /// - Shard splitting: After a shard split, ancestor layers beyond pitr_interval, so that
     413              :     ///   we don't indefinitely retain keys in this shard that aren't needed.
     414              :     /// - For future use: layers beyond pitr_interval that are in formats we would
     415              :     ///   rather not maintain compatibility with indefinitely.
     416              :     ///
     417              :     /// Note: this phase may read and write many gigabytes of data: use rewrite_max to bound
     418              :     /// how much work it will try to do in each compaction pass.
     419            0 :     async fn compact_shard_ancestors(
     420            0 :         self: &Arc<Self>,
     421            0 :         rewrite_max: usize,
     422            0 :         ctx: &RequestContext,
     423            0 :     ) -> Result<(), CompactionError> {
     424            0 :         let mut drop_layers = Vec::new();
     425            0 :         let mut layers_to_rewrite: Vec<Layer> = Vec::new();
     426            0 : 
     427            0 :         // We will use the Lsn cutoff of the last GC as a threshold for rewriting layers: if a
     428            0 :         // layer is behind this Lsn, it indicates that the layer is being retained beyond the
     429            0 :         // pitr_interval, for example because a branchpoint references it.
     430            0 :         //
     431            0 :         // Holding this read guard also blocks [`Self::gc_timeline`] from entering while we
     432            0 :         // are rewriting layers.
     433            0 :         let latest_gc_cutoff = self.get_latest_gc_cutoff_lsn();
     434            0 : 
     435            0 :         tracing::info!(
     436            0 :             "latest_gc_cutoff: {}, pitr cutoff {}",
     437            0 :             *latest_gc_cutoff,
     438            0 :             self.gc_info.read().unwrap().cutoffs.time
     439              :         );
     440              : 
     441            0 :         let layers = self.layers.read().await;
     442            0 :         for layer_desc in layers.layer_map()?.iter_historic_layers() {
     443            0 :             let layer = layers.get_from_desc(&layer_desc);
     444            0 :             if layer.metadata().shard.shard_count == self.shard_identity.count {
     445              :                 // This layer does not belong to a historic ancestor, no need to re-image it.
     446            0 :                 continue;
     447            0 :             }
     448            0 : 
     449            0 :             // This layer was created on an ancestor shard: check if it contains any data for this shard.
     450            0 :             let sharded_range = ShardedRange::new(layer_desc.get_key_range(), &self.shard_identity);
     451            0 :             let layer_local_page_count = sharded_range.page_count();
     452            0 :             let layer_raw_page_count = ShardedRange::raw_size(&layer_desc.get_key_range());
     453            0 :             if layer_local_page_count == 0 {
     454              :                 // This ancestral layer only covers keys that belong to other shards.
     455              :                 // We include the full metadata in the log: if we had some critical bug that caused
     456              :                 // us to incorrectly drop layers, this would simplify manually debugging + reinstating those layers.
     457            0 :                 info!(%layer, old_metadata=?layer.metadata(),
     458            0 :                     "dropping layer after shard split, contains no keys for this shard.",
     459              :                 );
     460              : 
     461            0 :                 if cfg!(debug_assertions) {
     462              :                     // Expensive, exhaustive check of keys in this layer: this guards against ShardedRange's calculations being
     463              :                     // wrong.  If ShardedRange claims the local page count is zero, then no keys in this layer
     464              :                     // should be !is_key_disposable()
     465            0 :                     let range = layer_desc.get_key_range();
     466            0 :                     let mut key = range.start;
     467            0 :                     while key < range.end {
     468            0 :                         debug_assert!(self.shard_identity.is_key_disposable(&key));
     469            0 :                         key = key.next();
     470              :                     }
     471            0 :                 }
     472              : 
     473            0 :                 drop_layers.push(layer);
     474            0 :                 continue;
     475            0 :             } else if layer_local_page_count != u32::MAX
     476            0 :                 && layer_local_page_count == layer_raw_page_count
     477              :             {
     478            0 :                 debug!(%layer,
     479            0 :                     "layer is entirely shard local ({} keys), no need to filter it",
     480              :                     layer_local_page_count
     481              :                 );
     482            0 :                 continue;
     483            0 :             }
     484            0 : 
     485            0 :             // Don't bother re-writing a layer unless it will at least halve its size
     486            0 :             if layer_local_page_count != u32::MAX
     487            0 :                 && layer_local_page_count > layer_raw_page_count / 2
     488              :             {
     489            0 :                 debug!(%layer,
     490            0 :                     "layer is already mostly local ({}/{}), not rewriting",
     491              :                     layer_local_page_count,
     492              :                     layer_raw_page_count
     493              :                 );
     494            0 :             }
     495              : 
     496              :             // Don't bother re-writing a layer if it is within the PITR window: it will age-out eventually
     497              :             // without incurring the I/O cost of a rewrite.
     498            0 :             if layer_desc.get_lsn_range().end >= *latest_gc_cutoff {
     499            0 :                 debug!(%layer, "Skipping rewrite of layer still in GC window ({} >= {})",
     500            0 :                     layer_desc.get_lsn_range().end, *latest_gc_cutoff);
     501            0 :                 continue;
     502            0 :             }
     503            0 : 
     504            0 :             if layer_desc.is_delta() {
     505              :                 // We do not yet implement rewrite of delta layers
     506            0 :                 debug!(%layer, "Skipping rewrite of delta layer");
     507            0 :                 continue;
     508            0 :             }
     509            0 : 
     510            0 :             // Only rewrite layers if their generations differ.  This guarantees:
     511            0 :             //  - that local rewrite is safe, as local layer paths will differ between existing layer and rewritten one
     512            0 :             //  - that the layer is persistent in remote storage, as we only see old-generation'd layer via loading from remote storage
     513            0 :             if layer.metadata().generation == self.generation {
     514            0 :                 debug!(%layer, "Skipping rewrite, is not from old generation");
     515            0 :                 continue;
     516            0 :             }
     517            0 : 
     518            0 :             if layers_to_rewrite.len() >= rewrite_max {
     519            0 :                 tracing::info!(%layer, "Will rewrite layer on a future compaction, already rewrote {}",
     520            0 :                     layers_to_rewrite.len()
     521              :                 );
     522            0 :                 continue;
     523            0 :             }
     524            0 : 
     525            0 :             // Fall through: all our conditions for doing a rewrite passed.
     526            0 :             layers_to_rewrite.push(layer);
     527              :         }
     528              : 
     529              :         // Drop read lock on layer map before we start doing time-consuming I/O
     530            0 :         drop(layers);
     531            0 : 
     532            0 :         let mut replace_image_layers = Vec::new();
     533              : 
     534            0 :         for layer in layers_to_rewrite {
     535            0 :             tracing::info!(layer=%layer, "Rewriting layer after shard split...");
     536            0 :             let mut image_layer_writer = ImageLayerWriter::new(
     537            0 :                 self.conf,
     538            0 :                 self.timeline_id,
     539            0 :                 self.tenant_shard_id,
     540            0 :                 &layer.layer_desc().key_range,
     541            0 :                 layer.layer_desc().image_layer_lsn(),
     542            0 :                 ctx,
     543            0 :             )
     544            0 :             .await
     545            0 :             .map_err(CompactionError::Other)?;
     546              : 
     547              :             // Safety of layer rewrites:
     548              :             // - We are writing to a different local file path than we are reading from, so the old Layer
     549              :             //   cannot interfere with the new one.
     550              :             // - In the page cache, contents for a particular VirtualFile are stored with a file_id that
     551              :             //   is different for two layers with the same name (in `ImageLayerInner::new` we always
     552              :             //   acquire a fresh id from [`crate::page_cache::next_file_id`].  So readers do not risk
     553              :             //   reading the index from one layer file, and then data blocks from the rewritten layer file.
     554              :             // - Any readers that have a reference to the old layer will keep it alive until they are done
     555              :             //   with it. If they are trying to promote from remote storage, that will fail, but this is the same
     556              :             //   as for compaction generally: compaction is allowed to delete layers that readers might be trying to use.
     557              :             // - We do not run concurrently with other kinds of compaction, so the only layer map writes we race with are:
     558              :             //    - GC, which at worst witnesses us "undelete" a layer that they just deleted.
     559              :             //    - ingestion, which only inserts layers, therefore cannot collide with us.
     560            0 :             let resident = layer.download_and_keep_resident().await?;
     561              : 
     562            0 :             let keys_written = resident
     563            0 :                 .filter(&self.shard_identity, &mut image_layer_writer, ctx)
     564            0 :                 .await?;
     565              : 
     566            0 :             if keys_written > 0 {
     567            0 :                 let (desc, path) = image_layer_writer
     568            0 :                     .finish(ctx)
     569            0 :                     .await
     570            0 :                     .map_err(CompactionError::Other)?;
     571            0 :                 let new_layer = Layer::finish_creating(self.conf, self, desc, &path)
     572            0 :                     .map_err(CompactionError::Other)?;
     573            0 :                 tracing::info!(layer=%new_layer, "Rewrote layer, {} -> {} bytes",
     574            0 :                     layer.metadata().file_size,
     575            0 :                     new_layer.metadata().file_size);
     576              : 
     577            0 :                 replace_image_layers.push((layer, new_layer));
     578            0 :             } else {
     579            0 :                 // Drop the old layer.  Usually for this case we would already have noticed that
     580            0 :                 // the layer has no data for us with the ShardedRange check above, but
     581            0 :                 drop_layers.push(layer);
     582            0 :             }
     583              :         }
     584              : 
     585              :         // At this point, we have replaced local layer files with their rewritten form, but not yet uploaded
     586              :         // metadata to reflect that. If we restart here, the replaced layer files will look invalid (size mismatch
     587              :         // to remote index) and be removed. This is inefficient but safe.
     588            0 :         fail::fail_point!("compact-shard-ancestors-localonly");
     589            0 : 
     590            0 :         // Update the LayerMap so that readers will use the new layers, and enqueue it for writing to remote storage
     591            0 :         self.rewrite_layers(replace_image_layers, drop_layers)
     592            0 :             .await?;
     593              : 
     594            0 :         fail::fail_point!("compact-shard-ancestors-enqueued");
     595            0 : 
     596            0 :         // We wait for all uploads to complete before finishing this compaction stage.  This is not
     597            0 :         // necessary for correctness, but it simplifies testing, and avoids proceeding with another
     598            0 :         // Timeline's compaction while this timeline's uploads may be generating lots of disk I/O
     599            0 :         // load.
     600            0 :         match self.remote_client.wait_completion().await {
     601            0 :             Ok(()) => (),
     602            0 :             Err(WaitCompletionError::NotInitialized(ni)) => return Err(CompactionError::from(ni)),
     603              :             Err(WaitCompletionError::UploadQueueShutDownOrStopped) => {
     604            0 :                 return Err(CompactionError::ShuttingDown)
     605              :             }
     606              :         }
     607              : 
     608            0 :         fail::fail_point!("compact-shard-ancestors-persistent");
     609            0 : 
     610            0 :         Ok(())
     611            0 :     }
     612              : 
     613              :     /// Update the LayerVisibilityHint of layers covered by image layers, based on whether there is
     614              :     /// an image layer between them and the most recent readable LSN (branch point or tip of timeline).  The
     615              :     /// purpose of the visibility hint is to record which layers need to be available to service reads.
     616              :     ///
     617              :     /// The result may be used as an input to eviction and secondary downloads to de-prioritize layers
     618              :     /// that we know won't be needed for reads.
     619         2160 :     pub(super) async fn update_layer_visibility(
     620         2160 :         &self,
     621         2160 :     ) -> Result<(), super::layer_manager::Shutdown> {
     622         2160 :         let head_lsn = self.get_last_record_lsn();
     623              : 
     624              :         // We will sweep through layers in reverse-LSN order.  We only do historic layers.  L0 deltas
     625              :         // are implicitly left visible, because LayerVisibilityHint's default is Visible, and we never modify it here.
     626              :         // Note that L0 deltas _can_ be covered by image layers, but we consider them 'visible' because we anticipate that
     627              :         // they will be subject to L0->L1 compaction in the near future.
     628         2160 :         let layer_manager = self.layers.read().await;
     629         2160 :         let layer_map = layer_manager.layer_map()?;
     630              : 
     631         2160 :         let readable_points = {
     632         2160 :             let children = self.gc_info.read().unwrap().retain_lsns.clone();
     633         2160 : 
     634         2160 :             let mut readable_points = Vec::with_capacity(children.len() + 1);
     635         2160 :             for (child_lsn, _child_timeline_id) in &children {
     636            0 :                 readable_points.push(*child_lsn);
     637            0 :             }
     638         2160 :             readable_points.push(head_lsn);
     639         2160 :             readable_points
     640         2160 :         };
     641         2160 : 
     642         2160 :         let (layer_visibility, covered) = layer_map.get_visibility(readable_points);
     643        10242 :         for (layer_desc, visibility) in layer_visibility {
     644         8082 :             // FIXME: a more efficiency bulk zip() through the layers rather than NlogN getting each one
     645         8082 :             let layer = layer_manager.get_from_desc(&layer_desc);
     646         8082 :             layer.set_visibility(visibility);
     647         8082 :         }
     648              : 
     649              :         // TODO: publish our covered KeySpace to our parent, so that when they update their visibility, they can
     650              :         // avoid assuming that everything at a branch point is visible.
     651         2160 :         drop(covered);
     652         2160 :         Ok(())
     653         2160 :     }
     654              : 
     655              :     /// Collect a bunch of Level 0 layer files, and compact and reshuffle them as
     656              :     /// as Level 1 files. Returns whether the L0 layers are fully compacted.
     657         1092 :     async fn compact_level0(
     658         1092 :         self: &Arc<Self>,
     659         1092 :         target_file_size: u64,
     660         1092 :         ctx: &RequestContext,
     661         1092 :     ) -> Result<bool, CompactionError> {
     662              :         let CompactLevel0Phase1Result {
     663         1092 :             new_layers,
     664         1092 :             deltas_to_compact,
     665         1092 :             fully_compacted,
     666              :         } = {
     667         1092 :             let phase1_span = info_span!("compact_level0_phase1");
     668         1092 :             let ctx = ctx.attached_child();
     669         1092 :             let mut stats = CompactLevel0Phase1StatsBuilder {
     670         1092 :                 version: Some(2),
     671         1092 :                 tenant_id: Some(self.tenant_shard_id),
     672         1092 :                 timeline_id: Some(self.timeline_id),
     673         1092 :                 ..Default::default()
     674         1092 :             };
     675         1092 : 
     676         1092 :             let begin = tokio::time::Instant::now();
     677         1092 :             let phase1_layers_locked = self.layers.read().await;
     678         1092 :             let now = tokio::time::Instant::now();
     679         1092 :             stats.read_lock_acquisition_micros =
     680         1092 :                 DurationRecorder::Recorded(RecordedDuration(now - begin), now);
     681         1092 :             self.compact_level0_phase1(phase1_layers_locked, stats, target_file_size, &ctx)
     682         1092 :                 .instrument(phase1_span)
     683        29649 :                 .await?
     684              :         };
     685              : 
     686         1092 :         if new_layers.is_empty() && deltas_to_compact.is_empty() {
     687              :             // nothing to do
     688         1008 :             return Ok(true);
     689           84 :         }
     690           84 : 
     691           84 :         self.finish_compact_batch(&new_layers, &Vec::new(), &deltas_to_compact)
     692            3 :             .await?;
     693           84 :         Ok(fully_compacted)
     694         1092 :     }
     695              : 
     696              :     /// Level0 files first phase of compaction, explained in the [`Self::compact_legacy`] comment.
     697         1092 :     async fn compact_level0_phase1<'a>(
     698         1092 :         self: &'a Arc<Self>,
     699         1092 :         guard: tokio::sync::RwLockReadGuard<'a, LayerManager>,
     700         1092 :         mut stats: CompactLevel0Phase1StatsBuilder,
     701         1092 :         target_file_size: u64,
     702         1092 :         ctx: &RequestContext,
     703         1092 :     ) -> Result<CompactLevel0Phase1Result, CompactionError> {
     704         1092 :         stats.read_lock_held_spawn_blocking_startup_micros =
     705         1092 :             stats.read_lock_acquisition_micros.till_now(); // set by caller
     706         1092 :         let layers = guard.layer_map()?;
     707         1092 :         let level0_deltas = layers.level0_deltas();
     708         1092 :         stats.level0_deltas_count = Some(level0_deltas.len());
     709         1092 : 
     710         1092 :         // Only compact if enough layers have accumulated.
     711         1092 :         let threshold = self.get_compaction_threshold();
     712         1092 :         if level0_deltas.is_empty() || level0_deltas.len() < threshold {
     713         1008 :             debug!(
     714            0 :                 level0_deltas = level0_deltas.len(),
     715            0 :                 threshold, "too few deltas to compact"
     716              :             );
     717         1008 :             return Ok(CompactLevel0Phase1Result::default());
     718           84 :         }
     719           84 : 
     720           84 :         let mut level0_deltas = level0_deltas
     721           84 :             .iter()
     722         1206 :             .map(|x| guard.get_from_desc(x))
     723           84 :             .collect::<Vec<_>>();
     724           84 : 
     725           84 :         // Gather the files to compact in this iteration.
     726           84 :         //
     727           84 :         // Start with the oldest Level 0 delta file, and collect any other
     728           84 :         // level 0 files that form a contiguous sequence, such that the end
     729           84 :         // LSN of previous file matches the start LSN of the next file.
     730           84 :         //
     731           84 :         // Note that if the files don't form such a sequence, we might
     732           84 :         // "compact" just a single file. That's a bit pointless, but it allows
     733           84 :         // us to get rid of the level 0 file, and compact the other files on
     734           84 :         // the next iteration. This could probably made smarter, but such
     735           84 :         // "gaps" in the sequence of level 0 files should only happen in case
     736           84 :         // of a crash, partial download from cloud storage, or something like
     737           84 :         // that, so it's not a big deal in practice.
     738         2244 :         level0_deltas.sort_by_key(|l| l.layer_desc().lsn_range.start);
     739           84 :         let mut level0_deltas_iter = level0_deltas.iter();
     740           84 : 
     741           84 :         let first_level0_delta = level0_deltas_iter.next().unwrap();
     742           84 :         let mut prev_lsn_end = first_level0_delta.layer_desc().lsn_range.end;
     743           84 :         let mut deltas_to_compact = Vec::with_capacity(level0_deltas.len());
     744           84 : 
     745           84 :         // Accumulate the size of layers in `deltas_to_compact`
     746           84 :         let mut deltas_to_compact_bytes = 0;
     747           84 : 
     748           84 :         // Under normal circumstances, we will accumulate up to compaction_interval L0s of size
     749           84 :         // checkpoint_distance each.  To avoid edge cases using extra system resources, bound our
     750           84 :         // work in this function to only operate on this much delta data at once.
     751           84 :         //
     752           84 :         // Take the max of the configured value & the default, so that tests that configure tiny values
     753           84 :         // can still use a sensible amount of memory, but if a deployed system configures bigger values we
     754           84 :         // still let them compact a full stack of L0s in one go.
     755           84 :         let delta_size_limit = std::cmp::max(
     756           84 :             self.get_compaction_threshold(),
     757           84 :             DEFAULT_COMPACTION_THRESHOLD,
     758           84 :         ) as u64
     759           84 :             * std::cmp::max(self.get_checkpoint_distance(), DEFAULT_CHECKPOINT_DISTANCE);
     760           84 : 
     761           84 :         let mut fully_compacted = true;
     762           84 : 
     763           84 :         deltas_to_compact.push(first_level0_delta.download_and_keep_resident().await?);
     764         1206 :         for l in level0_deltas_iter {
     765         1122 :             let lsn_range = &l.layer_desc().lsn_range;
     766         1122 : 
     767         1122 :             if lsn_range.start != prev_lsn_end {
     768            0 :                 break;
     769         1122 :             }
     770         1122 :             deltas_to_compact.push(l.download_and_keep_resident().await?);
     771         1122 :             deltas_to_compact_bytes += l.metadata().file_size;
     772         1122 :             prev_lsn_end = lsn_range.end;
     773         1122 : 
     774         1122 :             if deltas_to_compact_bytes >= delta_size_limit {
     775            0 :                 info!(
     776            0 :                     l0_deltas_selected = deltas_to_compact.len(),
     777            0 :                     l0_deltas_total = level0_deltas.len(),
     778            0 :                     "L0 compaction picker hit max delta layer size limit: {}",
     779              :                     delta_size_limit
     780              :                 );
     781            0 :                 fully_compacted = false;
     782            0 : 
     783            0 :                 // Proceed with compaction, but only a subset of L0s
     784            0 :                 break;
     785         1122 :             }
     786              :         }
     787           84 :         let lsn_range = Range {
     788           84 :             start: deltas_to_compact
     789           84 :                 .first()
     790           84 :                 .unwrap()
     791           84 :                 .layer_desc()
     792           84 :                 .lsn_range
     793           84 :                 .start,
     794           84 :             end: deltas_to_compact.last().unwrap().layer_desc().lsn_range.end,
     795           84 :         };
     796           84 : 
     797           84 :         info!(
     798            0 :             "Starting Level0 compaction in LSN range {}-{} for {} layers ({} deltas in total)",
     799            0 :             lsn_range.start,
     800            0 :             lsn_range.end,
     801            0 :             deltas_to_compact.len(),
     802            0 :             level0_deltas.len()
     803              :         );
     804              : 
     805         1206 :         for l in deltas_to_compact.iter() {
     806         1206 :             info!("compact includes {l}");
     807              :         }
     808              : 
     809              :         // We don't need the original list of layers anymore. Drop it so that
     810              :         // we don't accidentally use it later in the function.
     811           84 :         drop(level0_deltas);
     812           84 : 
     813           84 :         stats.read_lock_held_prerequisites_micros = stats
     814           84 :             .read_lock_held_spawn_blocking_startup_micros
     815           84 :             .till_now();
     816              : 
     817              :         // TODO: replace with streaming k-merge
     818           84 :         let all_keys = {
     819           84 :             let mut all_keys = Vec::new();
     820         1206 :             for l in deltas_to_compact.iter() {
     821         1206 :                 if self.cancel.is_cancelled() {
     822            0 :                     return Err(CompactionError::ShuttingDown);
     823         1206 :                 }
     824         7083 :                 all_keys.extend(l.load_keys(ctx).await.map_err(CompactionError::Other)?);
     825              :             }
     826              :             // The current stdlib sorting implementation is designed in a way where it is
     827              :             // particularly fast where the slice is made up of sorted sub-ranges.
     828     13271400 :             all_keys.sort_by_key(|DeltaEntry { key, lsn, .. }| (*key, *lsn));
     829           84 :             all_keys
     830           84 :         };
     831           84 : 
     832           84 :         stats.read_lock_held_key_sort_micros = stats.read_lock_held_prerequisites_micros.till_now();
     833              : 
     834              :         // Determine N largest holes where N is number of compacted layers. The vec is sorted by key range start.
     835              :         //
     836              :         // A hole is a key range for which this compaction doesn't have any WAL records.
     837              :         // Our goal in this compaction iteration is to avoid creating L1s that, in terms of their key range,
     838              :         // cover the hole, but actually don't contain any WAL records for that key range.
     839              :         // The reason is that the mere stack of L1s (`count_deltas`) triggers image layer creation (`create_image_layers`).
     840              :         // That image layer creation would be useless for a hole range covered by L1s that don't contain any WAL records.
     841              :         //
     842              :         // The algorithm chooses holes as follows.
     843              :         // - Slide a 2-window over the keys in key orde to get the hole range (=distance between two keys).
     844              :         // - Filter: min threshold on range length
     845              :         // - Rank: by coverage size (=number of image layers required to reconstruct each key in the range for which we have any data)
     846              :         //
     847              :         // For more details, intuition, and some ASCII art see https://github.com/neondatabase/neon/pull/3597#discussion_r1112704451
     848              :         #[derive(PartialEq, Eq)]
     849              :         struct Hole {
     850              :             key_range: Range<Key>,
     851              :             coverage_size: usize,
     852              :         }
     853           84 :         let holes: Vec<Hole> = {
     854              :             use std::cmp::Ordering;
     855              :             impl Ord for Hole {
     856            0 :                 fn cmp(&self, other: &Self) -> Ordering {
     857            0 :                     self.coverage_size.cmp(&other.coverage_size).reverse()
     858            0 :                 }
     859              :             }
     860              :             impl PartialOrd for Hole {
     861            0 :                 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
     862            0 :                     Some(self.cmp(other))
     863            0 :                 }
     864              :             }
     865           84 :             let max_holes = deltas_to_compact.len();
     866           84 :             let last_record_lsn = self.get_last_record_lsn();
     867           84 :             let min_hole_range = (target_file_size / page_cache::PAGE_SZ as u64) as i128;
     868           84 :             let min_hole_coverage_size = 3; // TODO: something more flexible?
     869           84 :                                             // min-heap (reserve space for one more element added before eviction)
     870           84 :             let mut heap: BinaryHeap<Hole> = BinaryHeap::with_capacity(max_holes + 1);
     871           84 :             let mut prev: Option<Key> = None;
     872              : 
     873      6192114 :             for &DeltaEntry { key: next_key, .. } in all_keys.iter() {
     874      6192114 :                 if let Some(prev_key) = prev {
     875              :                     // just first fast filter, do not create hole entries for metadata keys. The last hole in the
     876              :                     // compaction is the gap between data key and metadata keys.
     877      6192030 :                     if next_key.to_i128() - prev_key.to_i128() >= min_hole_range
     878            0 :                         && !Key::is_metadata_key(&prev_key)
     879              :                     {
     880            0 :                         let key_range = prev_key..next_key;
     881            0 :                         // Measuring hole by just subtraction of i128 representation of key range boundaries
     882            0 :                         // has not so much sense, because largest holes will corresponds field1/field2 changes.
     883            0 :                         // But we are mostly interested to eliminate holes which cause generation of excessive image layers.
     884            0 :                         // That is why it is better to measure size of hole as number of covering image layers.
     885            0 :                         let coverage_size =
     886            0 :                             layers.image_coverage(&key_range, last_record_lsn).len();
     887            0 :                         if coverage_size >= min_hole_coverage_size {
     888            0 :                             heap.push(Hole {
     889            0 :                                 key_range,
     890            0 :                                 coverage_size,
     891            0 :                             });
     892            0 :                             if heap.len() > max_holes {
     893            0 :                                 heap.pop(); // remove smallest hole
     894            0 :                             }
     895            0 :                         }
     896      6192030 :                     }
     897           84 :                 }
     898      6192114 :                 prev = Some(next_key.next());
     899              :             }
     900           84 :             let mut holes = heap.into_vec();
     901           84 :             holes.sort_unstable_by_key(|hole| hole.key_range.start);
     902           84 :             holes
     903           84 :         };
     904           84 :         stats.read_lock_held_compute_holes_micros = stats.read_lock_held_key_sort_micros.till_now();
     905           84 :         drop_rlock(guard);
     906           84 : 
     907           84 :         if self.cancel.is_cancelled() {
     908            0 :             return Err(CompactionError::ShuttingDown);
     909           84 :         }
     910           84 : 
     911           84 :         stats.read_lock_drop_micros = stats.read_lock_held_compute_holes_micros.till_now();
     912              : 
     913              :         // This iterator walks through all key-value pairs from all the layers
     914              :         // we're compacting, in key, LSN order.
     915              :         // If there's both a Value::Image and Value::WalRecord for the same (key,lsn),
     916              :         // then the Value::Image is ordered before Value::WalRecord.
     917           84 :         let mut all_values_iter = {
     918           84 :             let mut deltas = Vec::with_capacity(deltas_to_compact.len());
     919         1206 :             for l in deltas_to_compact.iter() {
     920         1206 :                 let l = l.get_as_delta(ctx).await.map_err(CompactionError::Other)?;
     921         1206 :                 deltas.push(l);
     922              :             }
     923           84 :             MergeIterator::create(&deltas, &[], ctx)
     924           84 :         };
     925           84 : 
     926           84 :         // This iterator walks through all keys and is needed to calculate size used by each key
     927           84 :         let mut all_keys_iter = all_keys
     928           84 :             .iter()
     929      6192114 :             .map(|DeltaEntry { key, lsn, size, .. }| (*key, *lsn, *size))
     930      6192030 :             .coalesce(|mut prev, cur| {
     931      6192030 :                 // Coalesce keys that belong to the same key pair.
     932      6192030 :                 // This ensures that compaction doesn't put them
     933      6192030 :                 // into different layer files.
     934      6192030 :                 // Still limit this by the target file size,
     935      6192030 :                 // so that we keep the size of the files in
     936      6192030 :                 // check.
     937      6192030 :                 if prev.0 == cur.0 && prev.2 < target_file_size {
     938       120114 :                     prev.2 += cur.2;
     939       120114 :                     Ok(prev)
     940              :                 } else {
     941      6071916 :                     Err((prev, cur))
     942              :                 }
     943      6192030 :             });
     944           84 : 
     945           84 :         // Merge the contents of all the input delta layers into a new set
     946           84 :         // of delta layers, based on the current partitioning.
     947           84 :         //
     948           84 :         // We split the new delta layers on the key dimension. We iterate through the key space, and for each key, check if including the next key to the current output layer we're building would cause the layer to become too large. If so, dump the current output layer and start new one.
     949           84 :         // It's possible that there is a single key with so many page versions that storing all of them in a single layer file
     950           84 :         // would be too large. In that case, we also split on the LSN dimension.
     951           84 :         //
     952           84 :         // LSN
     953           84 :         //  ^
     954           84 :         //  |
     955           84 :         //  | +-----------+            +--+--+--+--+
     956           84 :         //  | |           |            |  |  |  |  |
     957           84 :         //  | +-----------+            |  |  |  |  |
     958           84 :         //  | |           |            |  |  |  |  |
     959           84 :         //  | +-----------+     ==>    |  |  |  |  |
     960           84 :         //  | |           |            |  |  |  |  |
     961           84 :         //  | +-----------+            |  |  |  |  |
     962           84 :         //  | |           |            |  |  |  |  |
     963           84 :         //  | +-----------+            +--+--+--+--+
     964           84 :         //  |
     965           84 :         //  +--------------> key
     966           84 :         //
     967           84 :         //
     968           84 :         // If one key (X) has a lot of page versions:
     969           84 :         //
     970           84 :         // LSN
     971           84 :         //  ^
     972           84 :         //  |                                 (X)
     973           84 :         //  | +-----------+            +--+--+--+--+
     974           84 :         //  | |           |            |  |  |  |  |
     975           84 :         //  | +-----------+            |  |  +--+  |
     976           84 :         //  | |           |            |  |  |  |  |
     977           84 :         //  | +-----------+     ==>    |  |  |  |  |
     978           84 :         //  | |           |            |  |  +--+  |
     979           84 :         //  | +-----------+            |  |  |  |  |
     980           84 :         //  | |           |            |  |  |  |  |
     981           84 :         //  | +-----------+            +--+--+--+--+
     982           84 :         //  |
     983           84 :         //  +--------------> key
     984           84 :         // TODO: this actually divides the layers into fixed-size chunks, not
     985           84 :         // based on the partitioning.
     986           84 :         //
     987           84 :         // TODO: we should also opportunistically materialize and
     988           84 :         // garbage collect what we can.
     989           84 :         let mut new_layers = Vec::new();
     990           84 :         let mut prev_key: Option<Key> = None;
     991           84 :         let mut writer: Option<DeltaLayerWriter> = None;
     992           84 :         let mut key_values_total_size = 0u64;
     993           84 :         let mut dup_start_lsn: Lsn = Lsn::INVALID; // start LSN of layer containing values of the single key
     994           84 :         let mut dup_end_lsn: Lsn = Lsn::INVALID; // end LSN of layer containing values of the single key
     995           84 :         let mut next_hole = 0; // index of next hole in holes vector
     996           84 : 
     997           84 :         let mut keys = 0;
     998              : 
     999      6192198 :         while let Some((key, lsn, value)) = all_values_iter
    1000      6192198 :             .next()
    1001        10248 :             .await
    1002      6192198 :             .map_err(CompactionError::Other)?
    1003              :         {
    1004      6192114 :             keys += 1;
    1005      6192114 : 
    1006      6192114 :             if keys % 32_768 == 0 && self.cancel.is_cancelled() {
    1007              :                 // avoid hitting the cancellation token on every key. in benches, we end up
    1008              :                 // shuffling an order of million keys per layer, this means we'll check it
    1009              :                 // around tens of times per layer.
    1010            0 :                 return Err(CompactionError::ShuttingDown);
    1011      6192114 :             }
    1012      6192114 : 
    1013      6192114 :             let same_key = prev_key.map_or(false, |prev_key| prev_key == key);
    1014      6192114 :             // We need to check key boundaries once we reach next key or end of layer with the same key
    1015      6192114 :             if !same_key || lsn == dup_end_lsn {
    1016      6072000 :                 let mut next_key_size = 0u64;
    1017      6072000 :                 let is_dup_layer = dup_end_lsn.is_valid();
    1018      6072000 :                 dup_start_lsn = Lsn::INVALID;
    1019      6072000 :                 if !same_key {
    1020      6072000 :                     dup_end_lsn = Lsn::INVALID;
    1021      6072000 :                 }
    1022              :                 // Determine size occupied by this key. We stop at next key or when size becomes larger than target_file_size
    1023      6072000 :                 for (next_key, next_lsn, next_size) in all_keys_iter.by_ref() {
    1024      6072000 :                     next_key_size = next_size;
    1025      6072000 :                     if key != next_key {
    1026      6071916 :                         if dup_end_lsn.is_valid() {
    1027            0 :                             // We are writting segment with duplicates:
    1028            0 :                             // place all remaining values of this key in separate segment
    1029            0 :                             dup_start_lsn = dup_end_lsn; // new segments starts where old stops
    1030            0 :                             dup_end_lsn = lsn_range.end; // there are no more values of this key till end of LSN range
    1031      6071916 :                         }
    1032      6071916 :                         break;
    1033           84 :                     }
    1034           84 :                     key_values_total_size += next_size;
    1035           84 :                     // Check if it is time to split segment: if total keys size is larger than target file size.
    1036           84 :                     // We need to avoid generation of empty segments if next_size > target_file_size.
    1037           84 :                     if key_values_total_size > target_file_size && lsn != next_lsn {
    1038              :                         // Split key between multiple layers: such layer can contain only single key
    1039            0 :                         dup_start_lsn = if dup_end_lsn.is_valid() {
    1040            0 :                             dup_end_lsn // new segment with duplicates starts where old one stops
    1041              :                         } else {
    1042            0 :                             lsn // start with the first LSN for this key
    1043              :                         };
    1044            0 :                         dup_end_lsn = next_lsn; // upper LSN boundary is exclusive
    1045            0 :                         break;
    1046           84 :                     }
    1047              :                 }
    1048              :                 // handle case when loop reaches last key: in this case dup_end is non-zero but dup_start is not set.
    1049      6072000 :                 if dup_end_lsn.is_valid() && !dup_start_lsn.is_valid() {
    1050            0 :                     dup_start_lsn = dup_end_lsn;
    1051            0 :                     dup_end_lsn = lsn_range.end;
    1052      6072000 :                 }
    1053      6072000 :                 if writer.is_some() {
    1054      6071916 :                     let written_size = writer.as_mut().unwrap().size();
    1055      6071916 :                     let contains_hole =
    1056      6071916 :                         next_hole < holes.len() && key >= holes[next_hole].key_range.end;
    1057              :                     // check if key cause layer overflow or contains hole...
    1058      6071916 :                     if is_dup_layer
    1059      6071916 :                         || dup_end_lsn.is_valid()
    1060      6071916 :                         || written_size + key_values_total_size > target_file_size
    1061      6071076 :                         || contains_hole
    1062              :                     {
    1063              :                         // ... if so, flush previous layer and prepare to write new one
    1064          840 :                         let (desc, path) = writer
    1065          840 :                             .take()
    1066          840 :                             .unwrap()
    1067          840 :                             .finish(prev_key.unwrap().next(), ctx)
    1068         2130 :                             .await
    1069          840 :                             .map_err(CompactionError::Other)?;
    1070          840 :                         let new_delta = Layer::finish_creating(self.conf, self, desc, &path)
    1071          840 :                             .map_err(CompactionError::Other)?;
    1072              : 
    1073          840 :                         new_layers.push(new_delta);
    1074          840 :                         writer = None;
    1075          840 : 
    1076          840 :                         if contains_hole {
    1077            0 :                             // skip hole
    1078            0 :                             next_hole += 1;
    1079          840 :                         }
    1080      6071076 :                     }
    1081           84 :                 }
    1082              :                 // Remember size of key value because at next iteration we will access next item
    1083      6072000 :                 key_values_total_size = next_key_size;
    1084       120114 :             }
    1085      6192114 :             fail_point!("delta-layer-writer-fail-before-finish", |_| {
    1086            0 :                 Err(CompactionError::Other(anyhow::anyhow!(
    1087            0 :                     "failpoint delta-layer-writer-fail-before-finish"
    1088            0 :                 )))
    1089      6192114 :             });
    1090              : 
    1091      6192114 :             if !self.shard_identity.is_key_disposable(&key) {
    1092      6192114 :                 if writer.is_none() {
    1093          924 :                     if self.cancel.is_cancelled() {
    1094              :                         // to be somewhat responsive to cancellation, check for each new layer
    1095            0 :                         return Err(CompactionError::ShuttingDown);
    1096          924 :                     }
    1097              :                     // Create writer if not initiaized yet
    1098          924 :                     writer = Some(
    1099              :                         DeltaLayerWriter::new(
    1100          924 :                             self.conf,
    1101          924 :                             self.timeline_id,
    1102          924 :                             self.tenant_shard_id,
    1103          924 :                             key,
    1104          924 :                             if dup_end_lsn.is_valid() {
    1105              :                                 // this is a layer containing slice of values of the same key
    1106            0 :                                 debug!("Create new dup layer {}..{}", dup_start_lsn, dup_end_lsn);
    1107            0 :                                 dup_start_lsn..dup_end_lsn
    1108              :                             } else {
    1109          924 :                                 debug!("Create new layer {}..{}", lsn_range.start, lsn_range.end);
    1110          924 :                                 lsn_range.clone()
    1111              :                             },
    1112          924 :                             ctx,
    1113              :                         )
    1114          462 :                         .await
    1115          924 :                         .map_err(CompactionError::Other)?,
    1116              :                     );
    1117              : 
    1118          924 :                     keys = 0;
    1119      6191190 :                 }
    1120              : 
    1121      6192114 :                 writer
    1122      6192114 :                     .as_mut()
    1123      6192114 :                     .unwrap()
    1124      6192114 :                     .put_value(key, lsn, value, ctx)
    1125         3675 :                     .await
    1126      6192114 :                     .map_err(CompactionError::Other)?;
    1127              :             } else {
    1128            0 :                 debug!(
    1129            0 :                     "Dropping key {} during compaction (it belongs on shard {:?})",
    1130            0 :                     key,
    1131            0 :                     self.shard_identity.get_shard_number(&key)
    1132              :                 );
    1133              :             }
    1134              : 
    1135      6192114 :             if !new_layers.is_empty() {
    1136        59358 :                 fail_point!("after-timeline-compacted-first-L1");
    1137      6132756 :             }
    1138              : 
    1139      6192114 :             prev_key = Some(key);
    1140              :         }
    1141           84 :         if let Some(writer) = writer {
    1142           84 :             let (desc, path) = writer
    1143           84 :                 .finish(prev_key.unwrap().next(), ctx)
    1144         5967 :                 .await
    1145           84 :                 .map_err(CompactionError::Other)?;
    1146           84 :             let new_delta = Layer::finish_creating(self.conf, self, desc, &path)
    1147           84 :                 .map_err(CompactionError::Other)?;
    1148           84 :             new_layers.push(new_delta);
    1149            0 :         }
    1150              : 
    1151              :         // Sync layers
    1152           84 :         if !new_layers.is_empty() {
    1153              :             // Print a warning if the created layer is larger than double the target size
    1154              :             // Add two pages for potential overhead. This should in theory be already
    1155              :             // accounted for in the target calculation, but for very small targets,
    1156              :             // we still might easily hit the limit otherwise.
    1157           84 :             let warn_limit = target_file_size * 2 + page_cache::PAGE_SZ as u64 * 2;
    1158          924 :             for layer in new_layers.iter() {
    1159          924 :                 if layer.layer_desc().file_size > warn_limit {
    1160            0 :                     warn!(
    1161              :                         %layer,
    1162            0 :                         "created delta file of size {} larger than double of target of {target_file_size}", layer.layer_desc().file_size
    1163              :                     );
    1164          924 :                 }
    1165              :             }
    1166              : 
    1167              :             // The writer.finish() above already did the fsync of the inodes.
    1168              :             // We just need to fsync the directory in which these inodes are linked,
    1169              :             // which we know to be the timeline directory.
    1170              :             //
    1171              :             // We use fatal_err() below because the after writer.finish() returns with success,
    1172              :             // the in-memory state of the filesystem already has the layer file in its final place,
    1173              :             // and subsequent pageserver code could think it's durable while it really isn't.
    1174           84 :             let timeline_dir = VirtualFile::open(
    1175           84 :                 &self
    1176           84 :                     .conf
    1177           84 :                     .timeline_path(&self.tenant_shard_id, &self.timeline_id),
    1178           84 :                 ctx,
    1179           84 :             )
    1180           42 :             .await
    1181           84 :             .fatal_err("VirtualFile::open for timeline dir fsync");
    1182           84 :             timeline_dir
    1183           84 :                 .sync_all()
    1184           42 :                 .await
    1185           84 :                 .fatal_err("VirtualFile::sync_all timeline dir");
    1186            0 :         }
    1187              : 
    1188           84 :         stats.write_layer_files_micros = stats.read_lock_drop_micros.till_now();
    1189           84 :         stats.new_deltas_count = Some(new_layers.len());
    1190          924 :         stats.new_deltas_size = Some(new_layers.iter().map(|l| l.layer_desc().file_size).sum());
    1191           84 : 
    1192           84 :         match TryInto::<CompactLevel0Phase1Stats>::try_into(stats)
    1193           84 :             .and_then(|stats| serde_json::to_string(&stats).context("serde_json::to_string"))
    1194              :         {
    1195           84 :             Ok(stats_json) => {
    1196           84 :                 info!(
    1197            0 :                     stats_json = stats_json.as_str(),
    1198            0 :                     "compact_level0_phase1 stats available"
    1199              :                 )
    1200              :             }
    1201            0 :             Err(e) => {
    1202            0 :                 warn!("compact_level0_phase1 stats failed to serialize: {:#}", e);
    1203              :             }
    1204              :         }
    1205              : 
    1206              :         // Without this, rustc complains about deltas_to_compact still
    1207              :         // being borrowed when we `.into_iter()` below.
    1208           84 :         drop(all_values_iter);
    1209           84 : 
    1210           84 :         Ok(CompactLevel0Phase1Result {
    1211           84 :             new_layers,
    1212           84 :             deltas_to_compact: deltas_to_compact
    1213           84 :                 .into_iter()
    1214         1206 :                 .map(|x| x.drop_eviction_guard())
    1215           84 :                 .collect::<Vec<_>>(),
    1216           84 :             fully_compacted,
    1217           84 :         })
    1218         1092 :     }
    1219              : }
    1220              : 
    1221              : #[derive(Default)]
    1222              : struct CompactLevel0Phase1Result {
    1223              :     new_layers: Vec<ResidentLayer>,
    1224              :     deltas_to_compact: Vec<Layer>,
    1225              :     // Whether we have included all L0 layers, or selected only part of them due to the
    1226              :     // L0 compaction size limit.
    1227              :     fully_compacted: bool,
    1228              : }
    1229              : 
    1230              : #[derive(Default)]
    1231              : struct CompactLevel0Phase1StatsBuilder {
    1232              :     version: Option<u64>,
    1233              :     tenant_id: Option<TenantShardId>,
    1234              :     timeline_id: Option<TimelineId>,
    1235              :     read_lock_acquisition_micros: DurationRecorder,
    1236              :     read_lock_held_spawn_blocking_startup_micros: DurationRecorder,
    1237              :     read_lock_held_key_sort_micros: DurationRecorder,
    1238              :     read_lock_held_prerequisites_micros: DurationRecorder,
    1239              :     read_lock_held_compute_holes_micros: DurationRecorder,
    1240              :     read_lock_drop_micros: DurationRecorder,
    1241              :     write_layer_files_micros: DurationRecorder,
    1242              :     level0_deltas_count: Option<usize>,
    1243              :     new_deltas_count: Option<usize>,
    1244              :     new_deltas_size: Option<u64>,
    1245              : }
    1246              : 
    1247              : #[derive(serde::Serialize)]
    1248              : struct CompactLevel0Phase1Stats {
    1249              :     version: u64,
    1250              :     tenant_id: TenantShardId,
    1251              :     timeline_id: TimelineId,
    1252              :     read_lock_acquisition_micros: RecordedDuration,
    1253              :     read_lock_held_spawn_blocking_startup_micros: RecordedDuration,
    1254              :     read_lock_held_key_sort_micros: RecordedDuration,
    1255              :     read_lock_held_prerequisites_micros: RecordedDuration,
    1256              :     read_lock_held_compute_holes_micros: RecordedDuration,
    1257              :     read_lock_drop_micros: RecordedDuration,
    1258              :     write_layer_files_micros: RecordedDuration,
    1259              :     level0_deltas_count: usize,
    1260              :     new_deltas_count: usize,
    1261              :     new_deltas_size: u64,
    1262              : }
    1263              : 
    1264              : impl TryFrom<CompactLevel0Phase1StatsBuilder> for CompactLevel0Phase1Stats {
    1265              :     type Error = anyhow::Error;
    1266              : 
    1267           84 :     fn try_from(value: CompactLevel0Phase1StatsBuilder) -> Result<Self, Self::Error> {
    1268           84 :         Ok(Self {
    1269           84 :             version: value.version.ok_or_else(|| anyhow!("version not set"))?,
    1270           84 :             tenant_id: value
    1271           84 :                 .tenant_id
    1272           84 :                 .ok_or_else(|| anyhow!("tenant_id not set"))?,
    1273           84 :             timeline_id: value
    1274           84 :                 .timeline_id
    1275           84 :                 .ok_or_else(|| anyhow!("timeline_id not set"))?,
    1276           84 :             read_lock_acquisition_micros: value
    1277           84 :                 .read_lock_acquisition_micros
    1278           84 :                 .into_recorded()
    1279           84 :                 .ok_or_else(|| anyhow!("read_lock_acquisition_micros not set"))?,
    1280           84 :             read_lock_held_spawn_blocking_startup_micros: value
    1281           84 :                 .read_lock_held_spawn_blocking_startup_micros
    1282           84 :                 .into_recorded()
    1283           84 :                 .ok_or_else(|| anyhow!("read_lock_held_spawn_blocking_startup_micros not set"))?,
    1284           84 :             read_lock_held_key_sort_micros: value
    1285           84 :                 .read_lock_held_key_sort_micros
    1286           84 :                 .into_recorded()
    1287           84 :                 .ok_or_else(|| anyhow!("read_lock_held_key_sort_micros not set"))?,
    1288           84 :             read_lock_held_prerequisites_micros: value
    1289           84 :                 .read_lock_held_prerequisites_micros
    1290           84 :                 .into_recorded()
    1291           84 :                 .ok_or_else(|| anyhow!("read_lock_held_prerequisites_micros not set"))?,
    1292           84 :             read_lock_held_compute_holes_micros: value
    1293           84 :                 .read_lock_held_compute_holes_micros
    1294           84 :                 .into_recorded()
    1295           84 :                 .ok_or_else(|| anyhow!("read_lock_held_compute_holes_micros not set"))?,
    1296           84 :             read_lock_drop_micros: value
    1297           84 :                 .read_lock_drop_micros
    1298           84 :                 .into_recorded()
    1299           84 :                 .ok_or_else(|| anyhow!("read_lock_drop_micros not set"))?,
    1300           84 :             write_layer_files_micros: value
    1301           84 :                 .write_layer_files_micros
    1302           84 :                 .into_recorded()
    1303           84 :                 .ok_or_else(|| anyhow!("write_layer_files_micros not set"))?,
    1304           84 :             level0_deltas_count: value
    1305           84 :                 .level0_deltas_count
    1306           84 :                 .ok_or_else(|| anyhow!("level0_deltas_count not set"))?,
    1307           84 :             new_deltas_count: value
    1308           84 :                 .new_deltas_count
    1309           84 :                 .ok_or_else(|| anyhow!("new_deltas_count not set"))?,
    1310           84 :             new_deltas_size: value
    1311           84 :                 .new_deltas_size
    1312           84 :                 .ok_or_else(|| anyhow!("new_deltas_size not set"))?,
    1313              :         })
    1314           84 :     }
    1315              : }
    1316              : 
    1317              : impl Timeline {
    1318              :     /// Entry point for new tiered compaction algorithm.
    1319              :     ///
    1320              :     /// All the real work is in the implementation in the pageserver_compaction
    1321              :     /// crate. The code here would apply to any algorithm implemented by the
    1322              :     /// same interface, but tiered is the only one at the moment.
    1323              :     ///
    1324              :     /// TODO: cancellation
    1325            0 :     pub(crate) async fn compact_tiered(
    1326            0 :         self: &Arc<Self>,
    1327            0 :         _cancel: &CancellationToken,
    1328            0 :         ctx: &RequestContext,
    1329            0 :     ) -> Result<(), CompactionError> {
    1330            0 :         let fanout = self.get_compaction_threshold() as u64;
    1331            0 :         let target_file_size = self.get_checkpoint_distance();
    1332              : 
    1333              :         // Find the top of the historical layers
    1334            0 :         let end_lsn = {
    1335            0 :             let guard = self.layers.read().await;
    1336            0 :             let layers = guard.layer_map()?;
    1337              : 
    1338            0 :             let l0_deltas = layers.level0_deltas();
    1339            0 : 
    1340            0 :             // As an optimization, if we find that there are too few L0 layers,
    1341            0 :             // bail out early. We know that the compaction algorithm would do
    1342            0 :             // nothing in that case.
    1343            0 :             if l0_deltas.len() < fanout as usize {
    1344              :                 // doesn't need compacting
    1345            0 :                 return Ok(());
    1346            0 :             }
    1347            0 :             l0_deltas.iter().map(|l| l.lsn_range.end).max().unwrap()
    1348            0 :         };
    1349            0 : 
    1350            0 :         // Is the timeline being deleted?
    1351            0 :         if self.is_stopping() {
    1352            0 :             trace!("Dropping out of compaction on timeline shutdown");
    1353            0 :             return Err(CompactionError::ShuttingDown);
    1354            0 :         }
    1355              : 
    1356            0 :         let (dense_ks, _sparse_ks) = self.collect_keyspace(end_lsn, ctx).await?;
    1357              :         // TODO(chi): ignore sparse_keyspace for now, compact it in the future.
    1358            0 :         let mut adaptor = TimelineAdaptor::new(self, (end_lsn, dense_ks));
    1359            0 : 
    1360            0 :         pageserver_compaction::compact_tiered::compact_tiered(
    1361            0 :             &mut adaptor,
    1362            0 :             end_lsn,
    1363            0 :             target_file_size,
    1364            0 :             fanout,
    1365            0 :             ctx,
    1366            0 :         )
    1367            0 :         .await
    1368              :         // TODO: compact_tiered needs to return CompactionError
    1369            0 :         .map_err(CompactionError::Other)?;
    1370              : 
    1371            0 :         adaptor.flush_updates().await?;
    1372            0 :         Ok(())
    1373            0 :     }
    1374              : 
    1375              :     /// Take a list of images and deltas, produce images and deltas according to GC horizon and retain_lsns.
    1376              :     ///
    1377              :     /// It takes a key, the values of the key within the compaction process, a GC horizon, and all retain_lsns below the horizon.
    1378              :     /// For now, it requires the `accumulated_values` contains the full history of the key (i.e., the key with the lowest LSN is
    1379              :     /// an image or a WAL not requiring a base image). This restriction will be removed once we implement gc-compaction on branch.
    1380              :     ///
    1381              :     /// The function returns the deltas and the base image that need to be placed at each of the retain LSN. For example, we have:
    1382              :     ///
    1383              :     /// A@0x10, +B@0x20, +C@0x30, +D@0x40, +E@0x50, +F@0x60
    1384              :     /// horizon = 0x50, retain_lsn = 0x20, 0x40, delta_threshold=3
    1385              :     ///
    1386              :     /// The function will produce:
    1387              :     ///
    1388              :     /// ```plain
    1389              :     /// 0x20(retain_lsn) -> img=AB@0x20                  always produce a single image below the lowest retain LSN
    1390              :     /// 0x40(retain_lsn) -> deltas=[+C@0x30, +D@0x40]    two deltas since the last base image, keeping the deltas
    1391              :     /// 0x50(horizon)    -> deltas=[ABCDE@0x50]          three deltas since the last base image, generate an image but put it in the delta
    1392              :     /// above_horizon    -> deltas=[+F@0x60]             full history above the horizon
    1393              :     /// ```
    1394              :     ///
    1395              :     /// Note that `accumulated_values` must be sorted by LSN and should belong to a single key.
    1396         1290 :     pub(crate) async fn generate_key_retention(
    1397         1290 :         self: &Arc<Timeline>,
    1398         1290 :         key: Key,
    1399         1290 :         full_history: &[(Key, Lsn, Value)],
    1400         1290 :         horizon: Lsn,
    1401         1290 :         retain_lsn_below_horizon: &[Lsn],
    1402         1290 :         delta_threshold_cnt: usize,
    1403         1290 :         base_img_from_ancestor: Option<(Key, Lsn, Bytes)>,
    1404         1290 :     ) -> anyhow::Result<KeyHistoryRetention> {
    1405         1290 :         // Pre-checks for the invariants
    1406         1290 :         if cfg!(debug_assertions) {
    1407         3120 :             for (log_key, _, _) in full_history {
    1408         1830 :                 assert_eq!(log_key, &key, "mismatched key");
    1409              :             }
    1410         1290 :             for i in 1..full_history.len() {
    1411          540 :                 assert!(full_history[i - 1].1 <= full_history[i].1, "unordered LSN");
    1412          540 :                 if full_history[i - 1].1 == full_history[i].1 {
    1413            0 :                     assert!(
    1414            0 :                         matches!(full_history[i - 1].2, Value::Image(_)),
    1415            0 :                         "unordered delta/image, or duplicated delta"
    1416              :                     );
    1417          540 :                 }
    1418              :             }
    1419              :             // There was an assertion for no base image that checks if the first
    1420              :             // record in the history is `will_init` before, but it was removed.
    1421              :             // This is explained in the test cases for generate_key_retention.
    1422              :             // Search "incomplete history" for more information.
    1423         3000 :             for lsn in retain_lsn_below_horizon {
    1424         1710 :                 assert!(lsn < &horizon, "retain lsn must be below horizon")
    1425              :             }
    1426         1290 :             for i in 1..retain_lsn_below_horizon.len() {
    1427          834 :                 assert!(
    1428          834 :                     retain_lsn_below_horizon[i - 1] <= retain_lsn_below_horizon[i],
    1429            0 :                     "unordered LSN"
    1430              :                 );
    1431              :             }
    1432            0 :         }
    1433         1290 :         let has_ancestor = base_img_from_ancestor.is_some();
    1434              :         // Step 1: split history into len(retain_lsn_below_horizon) + 2 buckets, where the last bucket is for all deltas above the horizon,
    1435              :         // and the second-to-last bucket is for the horizon. Each bucket contains lsn_last_bucket < deltas <= lsn_this_bucket.
    1436         1290 :         let (mut split_history, lsn_split_points) = {
    1437         1290 :             let mut split_history = Vec::new();
    1438         1290 :             split_history.resize_with(retain_lsn_below_horizon.len() + 2, Vec::new);
    1439         1290 :             let mut lsn_split_points = Vec::with_capacity(retain_lsn_below_horizon.len() + 1);
    1440         3000 :             for lsn in retain_lsn_below_horizon {
    1441         1710 :                 lsn_split_points.push(*lsn);
    1442         1710 :             }
    1443         1290 :             lsn_split_points.push(horizon);
    1444         1290 :             let mut current_idx = 0;
    1445         3120 :             for item @ (_, lsn, _) in full_history {
    1446         2316 :                 while current_idx < lsn_split_points.len() && *lsn > lsn_split_points[current_idx] {
    1447          486 :                     current_idx += 1;
    1448          486 :                 }
    1449         1830 :                 split_history[current_idx].push(item);
    1450              :             }
    1451         1290 :             (split_history, lsn_split_points)
    1452              :         };
    1453              :         // Step 2: filter out duplicated records due to the k-merge of image/delta layers
    1454         5580 :         for split_for_lsn in &mut split_history {
    1455         4290 :             let mut prev_lsn = None;
    1456         4290 :             let mut new_split_for_lsn = Vec::with_capacity(split_for_lsn.len());
    1457         4290 :             for record @ (_, lsn, _) in std::mem::take(split_for_lsn) {
    1458         1830 :                 if let Some(prev_lsn) = &prev_lsn {
    1459          198 :                     if *prev_lsn == lsn {
    1460              :                         // The case that we have an LSN with both data from the delta layer and the image layer. As
    1461              :                         // `ValueWrapper` ensures that an image is ordered before a delta at the same LSN, we simply
    1462              :                         // drop this delta and keep the image.
    1463              :                         //
    1464              :                         // For example, we have delta layer key1@0x10, key1@0x20, and image layer key1@0x10, we will
    1465              :                         // keep the image for key1@0x10 and the delta for key1@0x20. key1@0x10 delta will be simply
    1466              :                         // dropped.
    1467              :                         //
    1468              :                         // TODO: in case we have both delta + images for a given LSN and it does not exceed the delta
    1469              :                         // threshold, we could have kept delta instead to save space. This is an optimization for the future.
    1470            0 :                         continue;
    1471          198 :                     }
    1472         1632 :                 }
    1473         1830 :                 prev_lsn = Some(lsn);
    1474         1830 :                 new_split_for_lsn.push(record);
    1475              :             }
    1476         4290 :             *split_for_lsn = new_split_for_lsn;
    1477              :         }
    1478              :         // Step 3: generate images when necessary
    1479         1290 :         let mut retention = Vec::with_capacity(split_history.len());
    1480         1290 :         let mut records_since_last_image = 0;
    1481         1290 :         let batch_cnt = split_history.len();
    1482         1290 :         assert!(
    1483         1290 :             batch_cnt >= 2,
    1484            0 :             "should have at least below + above horizon batches"
    1485              :         );
    1486         1290 :         let mut replay_history: Vec<(Key, Lsn, Value)> = Vec::new();
    1487         1290 :         if let Some((key, lsn, img)) = base_img_from_ancestor {
    1488           54 :             replay_history.push((key, lsn, Value::Image(img)));
    1489         1236 :         }
    1490              : 
    1491              :         /// Generate debug information for the replay history
    1492            0 :         fn generate_history_trace(replay_history: &[(Key, Lsn, Value)]) -> String {
    1493              :             use std::fmt::Write;
    1494            0 :             let mut output = String::new();
    1495            0 :             if let Some((key, _, _)) = replay_history.first() {
    1496            0 :                 write!(output, "key={} ", key).unwrap();
    1497            0 :                 let mut cnt = 0;
    1498            0 :                 for (_, lsn, val) in replay_history {
    1499            0 :                     if val.is_image() {
    1500            0 :                         write!(output, "i@{} ", lsn).unwrap();
    1501            0 :                     } else if val.will_init() {
    1502            0 :                         write!(output, "di@{} ", lsn).unwrap();
    1503            0 :                     } else {
    1504            0 :                         write!(output, "d@{} ", lsn).unwrap();
    1505            0 :                     }
    1506            0 :                     cnt += 1;
    1507            0 :                     if cnt >= 128 {
    1508            0 :                         write!(output, "... and more").unwrap();
    1509            0 :                         break;
    1510            0 :                     }
    1511              :                 }
    1512            0 :             } else {
    1513            0 :                 write!(output, "<no history>").unwrap();
    1514            0 :             }
    1515            0 :             output
    1516            0 :         }
    1517              : 
    1518            0 :         fn generate_debug_trace(
    1519            0 :             replay_history: Option<&[(Key, Lsn, Value)]>,
    1520            0 :             full_history: &[(Key, Lsn, Value)],
    1521            0 :             lsns: &[Lsn],
    1522            0 :             horizon: Lsn,
    1523            0 :         ) -> String {
    1524              :             use std::fmt::Write;
    1525            0 :             let mut output = String::new();
    1526            0 :             if let Some(replay_history) = replay_history {
    1527            0 :                 writeln!(
    1528            0 :                     output,
    1529            0 :                     "replay_history: {}",
    1530            0 :                     generate_history_trace(replay_history)
    1531            0 :                 )
    1532            0 :                 .unwrap();
    1533            0 :             } else {
    1534            0 :                 writeln!(output, "replay_history: <disabled>",).unwrap();
    1535            0 :             }
    1536            0 :             writeln!(
    1537            0 :                 output,
    1538            0 :                 "full_history: {}",
    1539            0 :                 generate_history_trace(full_history)
    1540            0 :             )
    1541            0 :             .unwrap();
    1542            0 :             writeln!(
    1543            0 :                 output,
    1544            0 :                 "when processing: [{}] horizon={}",
    1545            0 :                 lsns.iter().map(|l| format!("{l}")).join(","),
    1546            0 :                 horizon
    1547            0 :             )
    1548            0 :             .unwrap();
    1549            0 :             output
    1550            0 :         }
    1551              : 
    1552         4290 :         for (i, split_for_lsn) in split_history.into_iter().enumerate() {
    1553              :             // TODO: there could be image keys inside the splits, and we can compute records_since_last_image accordingly.
    1554         4290 :             records_since_last_image += split_for_lsn.len();
    1555         4290 :             let generate_image = if i == 0 && !has_ancestor {
    1556              :                 // We always generate images for the first batch (below horizon / lowest retain_lsn)
    1557         1236 :                 true
    1558         3054 :             } else if i == batch_cnt - 1 {
    1559              :                 // Do not generate images for the last batch (above horizon)
    1560         1290 :                 false
    1561         1764 :             } else if records_since_last_image >= delta_threshold_cnt {
    1562              :                 // Generate images when there are too many records
    1563           18 :                 true
    1564              :             } else {
    1565         1746 :                 false
    1566              :             };
    1567         4290 :             replay_history.extend(split_for_lsn.iter().map(|x| (*x).clone()));
    1568              :             // Only retain the items after the last image record
    1569         5274 :             for idx in (0..replay_history.len()).rev() {
    1570         5274 :                 if replay_history[idx].2.will_init() {
    1571         4290 :                     replay_history = replay_history[idx..].to_vec();
    1572         4290 :                     break;
    1573          984 :                 }
    1574              :             }
    1575         4290 :             if let Some((_, _, val)) = replay_history.first() {
    1576         4290 :                 if !val.will_init() {
    1577            0 :                     return Err(anyhow::anyhow!("invalid history, no base image")).with_context(
    1578            0 :                         || {
    1579            0 :                             generate_debug_trace(
    1580            0 :                                 Some(&replay_history),
    1581            0 :                                 full_history,
    1582            0 :                                 retain_lsn_below_horizon,
    1583            0 :                                 horizon,
    1584            0 :                             )
    1585            0 :                         },
    1586            0 :                     );
    1587         4290 :                 }
    1588            0 :             }
    1589         4290 :             if generate_image && records_since_last_image > 0 {
    1590         1254 :                 records_since_last_image = 0;
    1591         1254 :                 let replay_history_for_debug = if cfg!(debug_assertions) {
    1592         1254 :                     Some(replay_history.clone())
    1593              :                 } else {
    1594            0 :                     None
    1595              :                 };
    1596         1254 :                 let replay_history_for_debug_ref = replay_history_for_debug.as_deref();
    1597         1254 :                 let history = std::mem::take(&mut replay_history);
    1598         1254 :                 let mut img = None;
    1599         1254 :                 let mut records = Vec::with_capacity(history.len());
    1600         1254 :                 if let (_, lsn, Value::Image(val)) = history.first().as_ref().unwrap() {
    1601         1254 :                     img = Some((*lsn, val.clone()));
    1602         1254 :                     for (_, lsn, val) in history.into_iter().skip(1) {
    1603          102 :                         let Value::WalRecord(rec) = val else {
    1604            0 :                             return Err(anyhow::anyhow!(
    1605            0 :                                 "invalid record, first record is image, expect walrecords"
    1606            0 :                             ))
    1607            0 :                             .with_context(|| {
    1608            0 :                                 generate_debug_trace(
    1609            0 :                                     replay_history_for_debug_ref,
    1610            0 :                                     full_history,
    1611            0 :                                     retain_lsn_below_horizon,
    1612            0 :                                     horizon,
    1613            0 :                                 )
    1614            0 :                             });
    1615              :                         };
    1616          102 :                         records.push((lsn, rec));
    1617              :                     }
    1618              :                 } else {
    1619            0 :                     for (_, lsn, val) in history.into_iter() {
    1620            0 :                         let Value::WalRecord(rec) = val else {
    1621            0 :                             return Err(anyhow::anyhow!("invalid record, first record is walrecord, expect rest are walrecord"))
    1622            0 :                                 .with_context(|| generate_debug_trace(
    1623            0 :                                     replay_history_for_debug_ref,
    1624            0 :                                     full_history,
    1625            0 :                                     retain_lsn_below_horizon,
    1626            0 :                                     horizon,
    1627            0 :                                 ));
    1628              :                         };
    1629            0 :                         records.push((lsn, rec));
    1630              :                     }
    1631              :                 }
    1632         1254 :                 records.reverse();
    1633         1254 :                 let state = ValueReconstructState { img, records };
    1634         1254 :                 let request_lsn = lsn_split_points[i]; // last batch does not generate image so i is always in range
    1635         1254 :                 let img = self.reconstruct_value(key, request_lsn, state).await?;
    1636         1254 :                 replay_history.push((key, request_lsn, Value::Image(img.clone())));
    1637         1254 :                 retention.push(vec![(request_lsn, Value::Image(img))]);
    1638         3036 :             } else {
    1639         3036 :                 let deltas = split_for_lsn
    1640         3036 :                     .iter()
    1641         3036 :                     .map(|(_, lsn, value)| (*lsn, value.clone()))
    1642         3036 :                     .collect_vec();
    1643         3036 :                 retention.push(deltas);
    1644         3036 :             }
    1645              :         }
    1646         1290 :         let mut result = Vec::with_capacity(retention.len());
    1647         1290 :         assert_eq!(retention.len(), lsn_split_points.len() + 1);
    1648         4290 :         for (idx, logs) in retention.into_iter().enumerate() {
    1649         4290 :             if idx == lsn_split_points.len() {
    1650         1290 :                 return Ok(KeyHistoryRetention {
    1651         1290 :                     below_horizon: result,
    1652         1290 :                     above_horizon: KeyLogAtLsn(logs),
    1653         1290 :                 });
    1654         3000 :             } else {
    1655         3000 :                 result.push((lsn_split_points[idx], KeyLogAtLsn(logs)));
    1656         3000 :             }
    1657              :         }
    1658            0 :         unreachable!("key retention is empty")
    1659         1290 :     }
    1660              : 
    1661              :     /// An experimental compaction building block that combines compaction with garbage collection.
    1662              :     ///
    1663              :     /// The current implementation picks all delta + image layers that are below or intersecting with
    1664              :     /// the GC horizon without considering retain_lsns. Then, it does a full compaction over all these delta
    1665              :     /// layers and image layers, which generates image layers on the gc horizon, drop deltas below gc horizon,
    1666              :     /// and create delta layers with all deltas >= gc horizon.
    1667           78 :     pub(crate) async fn compact_with_gc(
    1668           78 :         self: &Arc<Self>,
    1669           78 :         cancel: &CancellationToken,
    1670           78 :         flags: EnumSet<CompactFlags>,
    1671           78 :         ctx: &RequestContext,
    1672           78 :     ) -> anyhow::Result<()> {
    1673              :         use std::collections::BTreeSet;
    1674              : 
    1675              :         // Block other compaction/GC tasks from running for now. GC-compaction could run along
    1676              :         // with legacy compaction tasks in the future. Always ensure the lock order is compaction -> gc.
    1677              :         // Note that we already acquired the compaction lock when the outer `compact` function gets called.
    1678              : 
    1679           78 :         let gc_lock = async {
    1680           78 :             tokio::select! {
    1681           78 :                 guard = self.gc_lock.lock() => Ok(guard),
    1682              :                 // TODO: refactor to CompactionError to correctly pass cancelled error
    1683           78 :                 _ = cancel.cancelled() => Err(anyhow!("cancelled")),
    1684              :             }
    1685           78 :         };
    1686              : 
    1687           78 :         let gc_lock = crate::timed(
    1688           78 :             gc_lock,
    1689           78 :             "acquires gc lock",
    1690           78 :             std::time::Duration::from_secs(5),
    1691           78 :         )
    1692            3 :         .await?;
    1693              : 
    1694           78 :         let dry_run = flags.contains(CompactFlags::DryRun);
    1695           78 : 
    1696           78 :         info!("running enhanced gc bottom-most compaction, dry_run={dry_run}");
    1697              : 
    1698           78 :         scopeguard::defer! {
    1699           78 :             info!("done enhanced gc bottom-most compaction");
    1700           78 :         };
    1701           78 : 
    1702           78 :         let mut stat = CompactionStatistics::default();
    1703              : 
    1704              :         // Step 0: pick all delta layers + image layers below/intersect with the GC horizon.
    1705              :         // The layer selection has the following properties:
    1706              :         // 1. If a layer is in the selection, all layers below it are in the selection.
    1707              :         // 2. Inferred from (1), for each key in the layer selection, the value can be reconstructed only with the layers in the layer selection.
    1708           78 :         let (layer_selection, gc_cutoff, retain_lsns_below_horizon) = {
    1709           78 :             let guard = self.layers.read().await;
    1710           78 :             let layers = guard.layer_map()?;
    1711           78 :             let gc_info = self.gc_info.read().unwrap();
    1712           78 :             let mut retain_lsns_below_horizon = Vec::new();
    1713           78 :             let gc_cutoff = gc_info.cutoffs.select_min();
    1714          102 :             for (lsn, _timeline_id) in &gc_info.retain_lsns {
    1715          102 :                 if lsn < &gc_cutoff {
    1716          102 :                     retain_lsns_below_horizon.push(*lsn);
    1717          102 :                 }
    1718              :             }
    1719           78 :             for lsn in gc_info.leases.keys() {
    1720            0 :                 if lsn < &gc_cutoff {
    1721            0 :                     retain_lsns_below_horizon.push(*lsn);
    1722            0 :                 }
    1723              :             }
    1724           78 :             let mut selected_layers = Vec::new();
    1725           78 :             drop(gc_info);
    1726              :             // Pick all the layers intersect or below the gc_cutoff, get the largest LSN in the selected layers.
    1727           78 :             let Some(max_layer_lsn) = layers
    1728           78 :                 .iter_historic_layers()
    1729          300 :                 .filter(|desc| desc.get_lsn_range().start <= gc_cutoff)
    1730          246 :                 .map(|desc| desc.get_lsn_range().end)
    1731           78 :                 .max()
    1732              :             else {
    1733            0 :                 info!("no layers to compact with gc");
    1734            0 :                 return Ok(());
    1735              :             };
    1736              :             // Then, pick all the layers that are below the max_layer_lsn. This is to ensure we can pick all single-key
    1737              :             // layers to compact.
    1738          300 :             for desc in layers.iter_historic_layers() {
    1739          300 :                 if desc.get_lsn_range().end <= max_layer_lsn {
    1740          246 :                     selected_layers.push(guard.get_from_desc(&desc));
    1741          246 :                 }
    1742              :             }
    1743           78 :             if selected_layers.is_empty() {
    1744            0 :                 info!("no layers to compact with gc");
    1745            0 :                 return Ok(());
    1746           78 :             }
    1747           78 :             retain_lsns_below_horizon.sort();
    1748           78 :             (selected_layers, gc_cutoff, retain_lsns_below_horizon)
    1749              :         };
    1750           78 :         let lowest_retain_lsn = if self.ancestor_timeline.is_some() {
    1751            6 :             Lsn(self.ancestor_lsn.0 + 1)
    1752              :         } else {
    1753           72 :             let res = retain_lsns_below_horizon
    1754           72 :                 .first()
    1755           72 :                 .copied()
    1756           72 :                 .unwrap_or(gc_cutoff);
    1757           72 :             if cfg!(debug_assertions) {
    1758           72 :                 assert_eq!(
    1759           72 :                     res,
    1760           72 :                     retain_lsns_below_horizon
    1761           72 :                         .iter()
    1762           72 :                         .min()
    1763           72 :                         .copied()
    1764           72 :                         .unwrap_or(gc_cutoff)
    1765           72 :                 );
    1766            0 :             }
    1767           72 :             res
    1768              :         };
    1769           78 :         info!(
    1770            0 :             "picked {} layers for compaction with gc_cutoff={} lowest_retain_lsn={}",
    1771            0 :             layer_selection.len(),
    1772              :             gc_cutoff,
    1773              :             lowest_retain_lsn
    1774              :         );
    1775              :         // Step 1: (In the future) construct a k-merge iterator over all layers. For now, simply collect all keys + LSNs.
    1776              :         // Also, verify if the layer map can be split by drawing a horizontal line at every LSN start/end split point.
    1777           78 :         let mut lsn_split_point = BTreeSet::new(); // TODO: use a better data structure (range tree / range set?)
    1778          324 :         for layer in &layer_selection {
    1779          246 :             let desc = layer.layer_desc();
    1780          246 :             if desc.is_delta() {
    1781              :                 // ignore single-key layer files
    1782          138 :                 if desc.key_range.start.next() != desc.key_range.end {
    1783          102 :                     let lsn_range = &desc.lsn_range;
    1784          102 :                     lsn_split_point.insert(lsn_range.start);
    1785          102 :                     lsn_split_point.insert(lsn_range.end);
    1786          102 :                 }
    1787          138 :                 stat.visit_delta_layer(desc.file_size());
    1788          108 :             } else {
    1789          108 :                 stat.visit_image_layer(desc.file_size());
    1790          108 :             }
    1791              :         }
    1792           78 :         let layer_names: Vec<crate::tenant::storage_layer::LayerName> = layer_selection
    1793           78 :             .iter()
    1794          246 :             .map(|layer| layer.layer_desc().layer_name())
    1795           78 :             .collect_vec();
    1796           78 :         if let Some(err) = check_valid_layermap(&layer_names) {
    1797            0 :             bail!("cannot run gc-compaction because {}", err);
    1798           78 :         }
    1799           78 :         // The maximum LSN we are processing in this compaction loop
    1800           78 :         let end_lsn = layer_selection
    1801           78 :             .iter()
    1802          246 :             .map(|l| l.layer_desc().lsn_range.end)
    1803           78 :             .max()
    1804           78 :             .unwrap();
    1805           78 :         // We don't want any of the produced layers to cover the full key range (i.e., MIN..MAX) b/c it will then be recognized
    1806           78 :         // as an L0 layer.
    1807           78 :         let mut delta_layers = Vec::new();
    1808           78 :         let mut image_layers = Vec::new();
    1809           78 :         let mut downloaded_layers = Vec::new();
    1810          324 :         for layer in &layer_selection {
    1811          246 :             let resident_layer = layer.download_and_keep_resident().await?;
    1812          246 :             downloaded_layers.push(resident_layer);
    1813              :         }
    1814          324 :         for resident_layer in &downloaded_layers {
    1815          246 :             if resident_layer.layer_desc().is_delta() {
    1816          138 :                 let layer = resident_layer.get_as_delta(ctx).await?;
    1817          138 :                 delta_layers.push(layer);
    1818              :             } else {
    1819          108 :                 let layer = resident_layer.get_as_image(ctx).await?;
    1820          108 :                 image_layers.push(layer);
    1821              :             }
    1822              :         }
    1823           78 :         let mut merge_iter = MergeIterator::create(&delta_layers, &image_layers, ctx);
    1824           78 :         // Step 2: Produce images+deltas. TODO: ensure newly-produced delta does not overlap with other deltas.
    1825           78 :         // Data of the same key.
    1826           78 :         let mut accumulated_values = Vec::new();
    1827           78 :         let mut last_key: Option<Key> = None;
    1828              : 
    1829              :         // Only create image layers when there is no ancestor branches. TODO: create covering image layer
    1830              :         // when some condition meet.
    1831           78 :         let mut image_layer_writer = if self.ancestor_timeline.is_none() {
    1832              :             Some(
    1833           72 :                 SplitImageLayerWriter::new(
    1834           72 :                     self.conf,
    1835           72 :                     self.timeline_id,
    1836           72 :                     self.tenant_shard_id,
    1837           72 :                     Key::MIN,
    1838           72 :                     lowest_retain_lsn,
    1839           72 :                     self.get_compaction_target_size(),
    1840           72 :                     ctx,
    1841           72 :                 )
    1842           36 :                 .await?,
    1843              :             )
    1844              :         } else {
    1845            6 :             None
    1846              :         };
    1847              : 
    1848           78 :         let mut delta_layer_writer = SplitDeltaLayerWriter::new(
    1849           78 :             self.conf,
    1850           78 :             self.timeline_id,
    1851           78 :             self.tenant_shard_id,
    1852           78 :             lowest_retain_lsn..end_lsn,
    1853           78 :             self.get_compaction_target_size(),
    1854           78 :         )
    1855            0 :         .await?;
    1856              : 
    1857              :         /// Returns None if there is no ancestor branch. Throw an error when the key is not found.
    1858              :         ///
    1859              :         /// Currently, we always get the ancestor image for each key in the child branch no matter whether the image
    1860              :         /// is needed for reconstruction. This should be fixed in the future.
    1861              :         ///
    1862              :         /// Furthermore, we should do vectored get instead of a single get, or better, use k-merge for ancestor
    1863              :         /// images.
    1864         1266 :         async fn get_ancestor_image(
    1865         1266 :             tline: &Arc<Timeline>,
    1866         1266 :             key: Key,
    1867         1266 :             ctx: &RequestContext,
    1868         1266 :         ) -> anyhow::Result<Option<(Key, Lsn, Bytes)>> {
    1869         1266 :             if tline.ancestor_timeline.is_none() {
    1870         1224 :                 return Ok(None);
    1871           42 :             };
    1872              :             // This function is implemented as a get of the current timeline at ancestor LSN, therefore reusing
    1873              :             // as much existing code as possible.
    1874           42 :             let img = tline.get(key, tline.ancestor_lsn, ctx).await?;
    1875           42 :             Ok(Some((key, tline.ancestor_lsn, img)))
    1876         1266 :         }
    1877              : 
    1878              :         // Actually, we can decide not to write to the image layer at all at this point because
    1879              :         // the key and LSN range are determined. However, to keep things simple here, we still
    1880              :         // create this writer, and discard the writer in the end.
    1881              : 
    1882         1758 :         while let Some((key, lsn, val)) = merge_iter.next().await? {
    1883         1680 :             if cancel.is_cancelled() {
    1884            0 :                 return Err(anyhow!("cancelled")); // TODO: refactor to CompactionError and pass cancel error
    1885         1680 :             }
    1886         1680 :             match val {
    1887         1260 :                 Value::Image(_) => stat.visit_image_key(&val),
    1888          420 :                 Value::WalRecord(_) => stat.visit_wal_key(&val),
    1889              :             }
    1890         1680 :             if last_key.is_none() || last_key.as_ref() == Some(&key) {
    1891          492 :                 if last_key.is_none() {
    1892           78 :                     last_key = Some(key);
    1893          414 :                 }
    1894          492 :                 accumulated_values.push((key, lsn, val));
    1895              :             } else {
    1896         1188 :                 let last_key = last_key.as_mut().unwrap();
    1897         1188 :                 stat.on_unique_key_visited();
    1898         1188 :                 let retention = self
    1899         1188 :                     .generate_key_retention(
    1900         1188 :                         *last_key,
    1901         1188 :                         &accumulated_values,
    1902         1188 :                         gc_cutoff,
    1903         1188 :                         &retain_lsns_below_horizon,
    1904         1188 :                         COMPACTION_DELTA_THRESHOLD,
    1905         1188 :                         get_ancestor_image(self, *last_key, ctx).await?,
    1906              :                     )
    1907            0 :                     .await?;
    1908              :                 // Put the image into the image layer. Currently we have a single big layer for the compaction.
    1909         1188 :                 retention
    1910         1188 :                     .pipe_to(
    1911         1188 :                         *last_key,
    1912         1188 :                         self,
    1913         1188 :                         &mut delta_layer_writer,
    1914         1188 :                         image_layer_writer.as_mut(),
    1915         1188 :                         &mut stat,
    1916         1188 :                         dry_run,
    1917         1188 :                         ctx,
    1918         1188 :                     )
    1919         1203 :                     .await?;
    1920         1188 :                 accumulated_values.clear();
    1921         1188 :                 *last_key = key;
    1922         1188 :                 accumulated_values.push((key, lsn, val));
    1923              :             }
    1924              :         }
    1925              : 
    1926           78 :         let last_key = last_key.expect("no keys produced during compaction");
    1927           78 :         // TODO: move this part to the loop body
    1928           78 :         stat.on_unique_key_visited();
    1929           78 :         let retention = self
    1930           78 :             .generate_key_retention(
    1931           78 :                 last_key,
    1932           78 :                 &accumulated_values,
    1933           78 :                 gc_cutoff,
    1934           78 :                 &retain_lsns_below_horizon,
    1935           78 :                 COMPACTION_DELTA_THRESHOLD,
    1936           78 :                 get_ancestor_image(self, last_key, ctx).await?,
    1937              :             )
    1938            0 :             .await?;
    1939              :         // Put the image into the image layer. Currently we have a single big layer for the compaction.
    1940           78 :         retention
    1941           78 :             .pipe_to(
    1942           78 :                 last_key,
    1943           78 :                 self,
    1944           78 :                 &mut delta_layer_writer,
    1945           78 :                 image_layer_writer.as_mut(),
    1946           78 :                 &mut stat,
    1947           78 :                 dry_run,
    1948           78 :                 ctx,
    1949           78 :             )
    1950           72 :             .await?;
    1951              : 
    1952          114 :         let discard = |key: &PersistentLayerKey| {
    1953          114 :             let key = key.clone();
    1954          114 :             async move { KeyHistoryRetention::discard_key(&key, self, dry_run).await }
    1955          114 :         };
    1956              : 
    1957           78 :         let produced_image_layers = if let Some(writer) = image_layer_writer {
    1958           72 :             if !dry_run {
    1959           60 :                 writer
    1960           60 :                     .finish_with_discard_fn(self, ctx, Key::MAX, discard)
    1961           72 :                     .await?
    1962              :             } else {
    1963           12 :                 let (layers, _) = writer.take()?;
    1964           12 :                 assert!(layers.is_empty(), "image layers produced in dry run mode?");
    1965           12 :                 Vec::new()
    1966              :             }
    1967              :         } else {
    1968            6 :             Vec::new()
    1969              :         };
    1970              : 
    1971           78 :         let produced_delta_layers = if !dry_run {
    1972           66 :             delta_layer_writer
    1973           66 :                 .finish_with_discard_fn(self, ctx, discard)
    1974           78 :                 .await?
    1975              :         } else {
    1976           12 :             let (layers, _) = delta_layer_writer.take()?;
    1977           12 :             assert!(layers.is_empty(), "delta layers produced in dry run mode?");
    1978           12 :             Vec::new()
    1979              :         };
    1980              : 
    1981           78 :         let mut compact_to = Vec::new();
    1982           78 :         let mut keep_layers = HashSet::new();
    1983           78 :         let produced_delta_layers_len = produced_delta_layers.len();
    1984           78 :         let produced_image_layers_len = produced_image_layers.len();
    1985          132 :         for action in produced_delta_layers {
    1986           54 :             match action {
    1987           30 :                 SplitWriterResult::Produced(layer) => {
    1988           30 :                     stat.produce_delta_layer(layer.layer_desc().file_size());
    1989           30 :                     compact_to.push(layer);
    1990           30 :                 }
    1991           24 :                 SplitWriterResult::Discarded(l) => {
    1992           24 :                     keep_layers.insert(l);
    1993           24 :                     stat.discard_delta_layer();
    1994           24 :                 }
    1995              :             }
    1996              :         }
    1997          138 :         for action in produced_image_layers {
    1998           60 :             match action {
    1999           36 :                 SplitWriterResult::Produced(layer) => {
    2000           36 :                     stat.produce_image_layer(layer.layer_desc().file_size());
    2001           36 :                     compact_to.push(layer);
    2002           36 :                 }
    2003           24 :                 SplitWriterResult::Discarded(l) => {
    2004           24 :                     keep_layers.insert(l);
    2005           24 :                     stat.discard_image_layer();
    2006           24 :                 }
    2007              :             }
    2008              :         }
    2009           78 :         let mut layer_selection = layer_selection;
    2010          246 :         layer_selection.retain(|x| !keep_layers.contains(&x.layer_desc().key()));
    2011           78 : 
    2012           78 :         info!(
    2013            0 :             "gc-compaction statistics: {}",
    2014            0 :             serde_json::to_string(&stat)?
    2015              :         );
    2016              : 
    2017           78 :         if dry_run {
    2018           12 :             return Ok(());
    2019           66 :         }
    2020           66 : 
    2021           66 :         info!(
    2022            0 :             "produced {} delta layers and {} image layers, {} layers are kept",
    2023            0 :             produced_delta_layers_len,
    2024            0 :             produced_image_layers_len,
    2025            0 :             layer_selection.len()
    2026              :         );
    2027              : 
    2028              :         // Step 3: Place back to the layer map.
    2029              :         {
    2030           66 :             let mut guard = self.layers.write().await;
    2031           66 :             guard
    2032           66 :                 .open_mut()?
    2033           66 :                 .finish_gc_compaction(&layer_selection, &compact_to, &self.metrics)
    2034           66 :         };
    2035           66 :         self.remote_client
    2036           66 :             .schedule_compaction_update(&layer_selection, &compact_to)?;
    2037              : 
    2038           66 :         drop(gc_lock);
    2039           66 : 
    2040           66 :         Ok(())
    2041           78 :     }
    2042              : }
    2043              : 
    2044              : struct TimelineAdaptor {
    2045              :     timeline: Arc<Timeline>,
    2046              : 
    2047              :     keyspace: (Lsn, KeySpace),
    2048              : 
    2049              :     new_deltas: Vec<ResidentLayer>,
    2050              :     new_images: Vec<ResidentLayer>,
    2051              :     layers_to_delete: Vec<Arc<PersistentLayerDesc>>,
    2052              : }
    2053              : 
    2054              : impl TimelineAdaptor {
    2055            0 :     pub fn new(timeline: &Arc<Timeline>, keyspace: (Lsn, KeySpace)) -> Self {
    2056            0 :         Self {
    2057            0 :             timeline: timeline.clone(),
    2058            0 :             keyspace,
    2059            0 :             new_images: Vec::new(),
    2060            0 :             new_deltas: Vec::new(),
    2061            0 :             layers_to_delete: Vec::new(),
    2062            0 :         }
    2063            0 :     }
    2064              : 
    2065            0 :     pub async fn flush_updates(&mut self) -> Result<(), CompactionError> {
    2066            0 :         let layers_to_delete = {
    2067            0 :             let guard = self.timeline.layers.read().await;
    2068            0 :             self.layers_to_delete
    2069            0 :                 .iter()
    2070            0 :                 .map(|x| guard.get_from_desc(x))
    2071            0 :                 .collect::<Vec<Layer>>()
    2072            0 :         };
    2073            0 :         self.timeline
    2074            0 :             .finish_compact_batch(&self.new_deltas, &self.new_images, &layers_to_delete)
    2075            0 :             .await?;
    2076              : 
    2077            0 :         self.timeline
    2078            0 :             .upload_new_image_layers(std::mem::take(&mut self.new_images))?;
    2079              : 
    2080            0 :         self.new_deltas.clear();
    2081            0 :         self.layers_to_delete.clear();
    2082            0 :         Ok(())
    2083            0 :     }
    2084              : }
    2085              : 
    2086              : #[derive(Clone)]
    2087              : struct ResidentDeltaLayer(ResidentLayer);
    2088              : #[derive(Clone)]
    2089              : struct ResidentImageLayer(ResidentLayer);
    2090              : 
    2091              : impl CompactionJobExecutor for TimelineAdaptor {
    2092              :     type Key = crate::repository::Key;
    2093              : 
    2094              :     type Layer = OwnArc<PersistentLayerDesc>;
    2095              :     type DeltaLayer = ResidentDeltaLayer;
    2096              :     type ImageLayer = ResidentImageLayer;
    2097              : 
    2098              :     type RequestContext = crate::context::RequestContext;
    2099              : 
    2100            0 :     fn get_shard_identity(&self) -> &ShardIdentity {
    2101            0 :         self.timeline.get_shard_identity()
    2102            0 :     }
    2103              : 
    2104            0 :     async fn get_layers(
    2105            0 :         &mut self,
    2106            0 :         key_range: &Range<Key>,
    2107            0 :         lsn_range: &Range<Lsn>,
    2108            0 :         _ctx: &RequestContext,
    2109            0 :     ) -> anyhow::Result<Vec<OwnArc<PersistentLayerDesc>>> {
    2110            0 :         self.flush_updates().await?;
    2111              : 
    2112            0 :         let guard = self.timeline.layers.read().await;
    2113            0 :         let layer_map = guard.layer_map()?;
    2114              : 
    2115            0 :         let result = layer_map
    2116            0 :             .iter_historic_layers()
    2117            0 :             .filter(|l| {
    2118            0 :                 overlaps_with(&l.lsn_range, lsn_range) && overlaps_with(&l.key_range, key_range)
    2119            0 :             })
    2120            0 :             .map(OwnArc)
    2121            0 :             .collect();
    2122            0 :         Ok(result)
    2123            0 :     }
    2124              : 
    2125            0 :     async fn get_keyspace(
    2126            0 :         &mut self,
    2127            0 :         key_range: &Range<Key>,
    2128            0 :         lsn: Lsn,
    2129            0 :         _ctx: &RequestContext,
    2130            0 :     ) -> anyhow::Result<Vec<Range<Key>>> {
    2131            0 :         if lsn == self.keyspace.0 {
    2132            0 :             Ok(pageserver_compaction::helpers::intersect_keyspace(
    2133            0 :                 &self.keyspace.1.ranges,
    2134            0 :                 key_range,
    2135            0 :             ))
    2136              :         } else {
    2137              :             // The current compaction implementation only ever requests the key space
    2138              :             // at the compaction end LSN.
    2139            0 :             anyhow::bail!("keyspace not available for requested lsn");
    2140              :         }
    2141            0 :     }
    2142              : 
    2143            0 :     async fn downcast_delta_layer(
    2144            0 :         &self,
    2145            0 :         layer: &OwnArc<PersistentLayerDesc>,
    2146            0 :     ) -> anyhow::Result<Option<ResidentDeltaLayer>> {
    2147            0 :         // this is a lot more complex than a simple downcast...
    2148            0 :         if layer.is_delta() {
    2149            0 :             let l = {
    2150            0 :                 let guard = self.timeline.layers.read().await;
    2151            0 :                 guard.get_from_desc(layer)
    2152              :             };
    2153            0 :             let result = l.download_and_keep_resident().await?;
    2154              : 
    2155            0 :             Ok(Some(ResidentDeltaLayer(result)))
    2156              :         } else {
    2157            0 :             Ok(None)
    2158              :         }
    2159            0 :     }
    2160              : 
    2161            0 :     async fn create_image(
    2162            0 :         &mut self,
    2163            0 :         lsn: Lsn,
    2164            0 :         key_range: &Range<Key>,
    2165            0 :         ctx: &RequestContext,
    2166            0 :     ) -> anyhow::Result<()> {
    2167            0 :         Ok(self.create_image_impl(lsn, key_range, ctx).await?)
    2168            0 :     }
    2169              : 
    2170            0 :     async fn create_delta(
    2171            0 :         &mut self,
    2172            0 :         lsn_range: &Range<Lsn>,
    2173            0 :         key_range: &Range<Key>,
    2174            0 :         input_layers: &[ResidentDeltaLayer],
    2175            0 :         ctx: &RequestContext,
    2176            0 :     ) -> anyhow::Result<()> {
    2177            0 :         debug!("Create new layer {}..{}", lsn_range.start, lsn_range.end);
    2178              : 
    2179            0 :         let mut all_entries = Vec::new();
    2180            0 :         for dl in input_layers.iter() {
    2181            0 :             all_entries.extend(dl.load_keys(ctx).await?);
    2182              :         }
    2183              : 
    2184              :         // The current stdlib sorting implementation is designed in a way where it is
    2185              :         // particularly fast where the slice is made up of sorted sub-ranges.
    2186            0 :         all_entries.sort_by_key(|DeltaEntry { key, lsn, .. }| (*key, *lsn));
    2187              : 
    2188            0 :         let mut writer = DeltaLayerWriter::new(
    2189            0 :             self.timeline.conf,
    2190            0 :             self.timeline.timeline_id,
    2191            0 :             self.timeline.tenant_shard_id,
    2192            0 :             key_range.start,
    2193            0 :             lsn_range.clone(),
    2194            0 :             ctx,
    2195            0 :         )
    2196            0 :         .await?;
    2197              : 
    2198            0 :         let mut dup_values = 0;
    2199            0 : 
    2200            0 :         // This iterator walks through all key-value pairs from all the layers
    2201            0 :         // we're compacting, in key, LSN order.
    2202            0 :         let mut prev: Option<(Key, Lsn)> = None;
    2203              :         for &DeltaEntry {
    2204            0 :             key, lsn, ref val, ..
    2205            0 :         } in all_entries.iter()
    2206              :         {
    2207            0 :             if prev == Some((key, lsn)) {
    2208              :                 // This is a duplicate. Skip it.
    2209              :                 //
    2210              :                 // It can happen if compaction is interrupted after writing some
    2211              :                 // layers but not all, and we are compacting the range again.
    2212              :                 // The calculations in the algorithm assume that there are no
    2213              :                 // duplicates, so the math on targeted file size is likely off,
    2214              :                 // and we will create smaller files than expected.
    2215            0 :                 dup_values += 1;
    2216            0 :                 continue;
    2217            0 :             }
    2218              : 
    2219            0 :             let value = val.load(ctx).await?;
    2220              : 
    2221            0 :             writer.put_value(key, lsn, value, ctx).await?;
    2222              : 
    2223            0 :             prev = Some((key, lsn));
    2224              :         }
    2225              : 
    2226            0 :         if dup_values > 0 {
    2227            0 :             warn!("delta layer created with {} duplicate values", dup_values);
    2228            0 :         }
    2229              : 
    2230            0 :         fail_point!("delta-layer-writer-fail-before-finish", |_| {
    2231            0 :             Err(anyhow::anyhow!(
    2232            0 :                 "failpoint delta-layer-writer-fail-before-finish"
    2233            0 :             ))
    2234            0 :         });
    2235              : 
    2236            0 :         let (desc, path) = writer.finish(prev.unwrap().0.next(), ctx).await?;
    2237            0 :         let new_delta_layer =
    2238            0 :             Layer::finish_creating(self.timeline.conf, &self.timeline, desc, &path)?;
    2239              : 
    2240            0 :         self.new_deltas.push(new_delta_layer);
    2241            0 :         Ok(())
    2242            0 :     }
    2243              : 
    2244            0 :     async fn delete_layer(
    2245            0 :         &mut self,
    2246            0 :         layer: &OwnArc<PersistentLayerDesc>,
    2247            0 :         _ctx: &RequestContext,
    2248            0 :     ) -> anyhow::Result<()> {
    2249            0 :         self.layers_to_delete.push(layer.clone().0);
    2250            0 :         Ok(())
    2251            0 :     }
    2252              : }
    2253              : 
    2254              : impl TimelineAdaptor {
    2255            0 :     async fn create_image_impl(
    2256            0 :         &mut self,
    2257            0 :         lsn: Lsn,
    2258            0 :         key_range: &Range<Key>,
    2259            0 :         ctx: &RequestContext,
    2260            0 :     ) -> Result<(), CreateImageLayersError> {
    2261            0 :         let timer = self.timeline.metrics.create_images_time_histo.start_timer();
    2262              : 
    2263            0 :         let image_layer_writer = ImageLayerWriter::new(
    2264            0 :             self.timeline.conf,
    2265            0 :             self.timeline.timeline_id,
    2266            0 :             self.timeline.tenant_shard_id,
    2267            0 :             key_range,
    2268            0 :             lsn,
    2269            0 :             ctx,
    2270            0 :         )
    2271            0 :         .await?;
    2272              : 
    2273            0 :         fail_point!("image-layer-writer-fail-before-finish", |_| {
    2274            0 :             Err(CreateImageLayersError::Other(anyhow::anyhow!(
    2275            0 :                 "failpoint image-layer-writer-fail-before-finish"
    2276            0 :             )))
    2277            0 :         });
    2278              : 
    2279            0 :         let keyspace = KeySpace {
    2280            0 :             ranges: self.get_keyspace(key_range, lsn, ctx).await?,
    2281              :         };
    2282              :         // TODO set proper (stateful) start. The create_image_layer_for_rel_blocks function mostly
    2283            0 :         let start = Key::MIN;
    2284              :         let ImageLayerCreationOutcome {
    2285            0 :             image,
    2286              :             next_start_key: _,
    2287            0 :         } = self
    2288            0 :             .timeline
    2289            0 :             .create_image_layer_for_rel_blocks(
    2290            0 :                 &keyspace,
    2291            0 :                 image_layer_writer,
    2292            0 :                 lsn,
    2293            0 :                 ctx,
    2294            0 :                 key_range.clone(),
    2295            0 :                 start,
    2296            0 :             )
    2297            0 :             .await?;
    2298              : 
    2299            0 :         if let Some(image_layer) = image {
    2300            0 :             self.new_images.push(image_layer);
    2301            0 :         }
    2302              : 
    2303            0 :         timer.stop_and_record();
    2304            0 : 
    2305            0 :         Ok(())
    2306            0 :     }
    2307              : }
    2308              : 
    2309              : impl CompactionRequestContext for crate::context::RequestContext {}
    2310              : 
    2311              : #[derive(Debug, Clone)]
    2312              : pub struct OwnArc<T>(pub Arc<T>);
    2313              : 
    2314              : impl<T> Deref for OwnArc<T> {
    2315              :     type Target = <Arc<T> as Deref>::Target;
    2316            0 :     fn deref(&self) -> &Self::Target {
    2317            0 :         &self.0
    2318            0 :     }
    2319              : }
    2320              : 
    2321              : impl<T> AsRef<T> for OwnArc<T> {
    2322            0 :     fn as_ref(&self) -> &T {
    2323            0 :         self.0.as_ref()
    2324            0 :     }
    2325              : }
    2326              : 
    2327              : impl CompactionLayer<Key> for OwnArc<PersistentLayerDesc> {
    2328            0 :     fn key_range(&self) -> &Range<Key> {
    2329            0 :         &self.key_range
    2330            0 :     }
    2331            0 :     fn lsn_range(&self) -> &Range<Lsn> {
    2332            0 :         &self.lsn_range
    2333            0 :     }
    2334            0 :     fn file_size(&self) -> u64 {
    2335            0 :         self.file_size
    2336            0 :     }
    2337            0 :     fn short_id(&self) -> std::string::String {
    2338            0 :         self.as_ref().short_id().to_string()
    2339            0 :     }
    2340            0 :     fn is_delta(&self) -> bool {
    2341            0 :         self.as_ref().is_delta()
    2342            0 :     }
    2343              : }
    2344              : 
    2345              : impl CompactionLayer<Key> for OwnArc<DeltaLayer> {
    2346            0 :     fn key_range(&self) -> &Range<Key> {
    2347            0 :         &self.layer_desc().key_range
    2348            0 :     }
    2349            0 :     fn lsn_range(&self) -> &Range<Lsn> {
    2350            0 :         &self.layer_desc().lsn_range
    2351            0 :     }
    2352            0 :     fn file_size(&self) -> u64 {
    2353            0 :         self.layer_desc().file_size
    2354            0 :     }
    2355            0 :     fn short_id(&self) -> std::string::String {
    2356            0 :         self.layer_desc().short_id().to_string()
    2357            0 :     }
    2358            0 :     fn is_delta(&self) -> bool {
    2359            0 :         true
    2360            0 :     }
    2361              : }
    2362              : 
    2363              : use crate::tenant::timeline::DeltaEntry;
    2364              : 
    2365              : impl CompactionLayer<Key> for ResidentDeltaLayer {
    2366            0 :     fn key_range(&self) -> &Range<Key> {
    2367            0 :         &self.0.layer_desc().key_range
    2368            0 :     }
    2369            0 :     fn lsn_range(&self) -> &Range<Lsn> {
    2370            0 :         &self.0.layer_desc().lsn_range
    2371            0 :     }
    2372            0 :     fn file_size(&self) -> u64 {
    2373            0 :         self.0.layer_desc().file_size
    2374            0 :     }
    2375            0 :     fn short_id(&self) -> std::string::String {
    2376            0 :         self.0.layer_desc().short_id().to_string()
    2377            0 :     }
    2378            0 :     fn is_delta(&self) -> bool {
    2379            0 :         true
    2380            0 :     }
    2381              : }
    2382              : 
    2383              : impl CompactionDeltaLayer<TimelineAdaptor> for ResidentDeltaLayer {
    2384              :     type DeltaEntry<'a> = DeltaEntry<'a>;
    2385              : 
    2386            0 :     async fn load_keys<'a>(&self, ctx: &RequestContext) -> anyhow::Result<Vec<DeltaEntry<'_>>> {
    2387            0 :         self.0.load_keys(ctx).await
    2388            0 :     }
    2389              : }
    2390              : 
    2391              : impl CompactionLayer<Key> for ResidentImageLayer {
    2392            0 :     fn key_range(&self) -> &Range<Key> {
    2393            0 :         &self.0.layer_desc().key_range
    2394            0 :     }
    2395            0 :     fn lsn_range(&self) -> &Range<Lsn> {
    2396            0 :         &self.0.layer_desc().lsn_range
    2397            0 :     }
    2398            0 :     fn file_size(&self) -> u64 {
    2399            0 :         self.0.layer_desc().file_size
    2400            0 :     }
    2401            0 :     fn short_id(&self) -> std::string::String {
    2402            0 :         self.0.layer_desc().short_id().to_string()
    2403            0 :     }
    2404            0 :     fn is_delta(&self) -> bool {
    2405            0 :         false
    2406            0 :     }
    2407              : }
    2408              : impl CompactionImageLayer<TimelineAdaptor> for ResidentImageLayer {}
        

Generated by: LCOV version 2.1-beta