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