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, VecDeque};
8 : use std::ops::{Deref, Range};
9 : use std::sync::Arc;
10 : use std::time::{Duration, Instant};
11 :
12 : use super::layer_manager::LayerManager;
13 : use super::{
14 : CompactFlags, CompactOptions, CompactionError, CreateImageLayersError, DurationRecorder,
15 : GetVectoredError, ImageLayerCreationMode, LastImageLayerCreationStatus, RecordedDuration,
16 : Timeline,
17 : };
18 :
19 : use crate::tenant::timeline::DeltaEntry;
20 : use crate::walredo::RedoAttemptType;
21 : use anyhow::{Context, anyhow};
22 : use bytes::Bytes;
23 : use enumset::EnumSet;
24 : use fail::fail_point;
25 : use futures::FutureExt;
26 : use itertools::Itertools;
27 : use once_cell::sync::Lazy;
28 : use pageserver_api::config::tenant_conf_defaults::DEFAULT_CHECKPOINT_DISTANCE;
29 : use pageserver_api::key::{KEY_SIZE, Key};
30 : use pageserver_api::keyspace::{KeySpace, ShardedRange};
31 : use pageserver_api::models::{CompactInfoResponse, CompactKeyRange};
32 : use pageserver_api::record::NeonWalRecord;
33 : use pageserver_api::shard::{ShardCount, ShardIdentity, TenantShardId};
34 : use pageserver_api::value::Value;
35 : use pageserver_compaction::helpers::{fully_contains, overlaps_with};
36 : use pageserver_compaction::interface::*;
37 : use serde::Serialize;
38 : use tokio::sync::{OwnedSemaphorePermit, Semaphore};
39 : use tokio_util::sync::CancellationToken;
40 : use tracing::{Instrument, debug, error, info, info_span, trace, warn};
41 : use utils::critical;
42 : use utils::id::TimelineId;
43 : use utils::lsn::Lsn;
44 :
45 : use crate::context::{AccessStatsBehavior, RequestContext, RequestContextBuilder};
46 : use crate::page_cache;
47 : use crate::statvfs::Statvfs;
48 : use crate::tenant::checks::check_valid_layermap;
49 : use crate::tenant::gc_block::GcBlock;
50 : use crate::tenant::layer_map::LayerMap;
51 : use crate::tenant::remote_timeline_client::WaitCompletionError;
52 : use crate::tenant::remote_timeline_client::index::GcCompactionState;
53 : use crate::tenant::storage_layer::batch_split_writer::{
54 : BatchWriterResult, SplitDeltaLayerWriter, SplitImageLayerWriter,
55 : };
56 : use crate::tenant::storage_layer::filter_iterator::FilterIterator;
57 : use crate::tenant::storage_layer::merge_iterator::MergeIterator;
58 : use crate::tenant::storage_layer::{
59 : AsLayerDesc, LayerVisibilityHint, PersistentLayerDesc, PersistentLayerKey,
60 : ValueReconstructState,
61 : };
62 : use crate::tenant::tasks::log_compaction_error;
63 : use crate::tenant::timeline::{
64 : DeltaLayerWriter, ImageLayerCreationOutcome, ImageLayerWriter, IoConcurrency, Layer,
65 : ResidentLayer, drop_rlock,
66 : };
67 : use crate::tenant::{DeltaLayer, MaybeOffloaded};
68 : use crate::virtual_file::{MaybeFatalIo, VirtualFile};
69 :
70 : /// Maximum number of deltas before generating an image layer in bottom-most compaction.
71 : const COMPACTION_DELTA_THRESHOLD: usize = 5;
72 :
73 : /// Ratio of shard-local pages below which we trigger shard ancestor layer rewrites. 0.3 means that
74 : /// <= 30% of layer pages must belong to the descendant shard to rewrite the layer.
75 : ///
76 : /// We choose a value < 0.5 to avoid rewriting all visible layers every time we do a power-of-two
77 : /// shard split, which gets expensive for large tenants.
78 : const ANCESTOR_COMPACTION_REWRITE_THRESHOLD: f64 = 0.3;
79 :
80 : #[derive(Default, Debug, Clone, Copy, Hash, PartialEq, Eq, Serialize)]
81 : pub struct GcCompactionJobId(pub usize);
82 :
83 : impl std::fmt::Display for GcCompactionJobId {
84 0 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
85 0 : write!(f, "{}", self.0)
86 0 : }
87 : }
88 :
89 : pub struct GcCompactionCombinedSettings {
90 : pub gc_compaction_enabled: bool,
91 : pub gc_compaction_verification: bool,
92 : pub gc_compaction_initial_threshold_kb: u64,
93 : pub gc_compaction_ratio_percent: u64,
94 : }
95 :
96 : #[derive(Debug, Clone)]
97 : pub enum GcCompactionQueueItem {
98 : MetaJob {
99 : /// Compaction options
100 : options: CompactOptions,
101 : /// Whether the compaction is triggered automatically (determines whether we need to update L2 LSN)
102 : auto: bool,
103 : },
104 : SubCompactionJob(CompactOptions),
105 : Notify(GcCompactionJobId, Option<Lsn>),
106 : }
107 :
108 : /// Statistics for gc-compaction meta jobs, which contains several sub compaction jobs.
109 : #[derive(Debug, Clone, Serialize, Default)]
110 : pub struct GcCompactionMetaStatistics {
111 : /// The total number of sub compaction jobs.
112 : pub total_sub_compaction_jobs: usize,
113 : /// The total number of sub compaction jobs that failed.
114 : pub failed_sub_compaction_jobs: usize,
115 : /// The total number of sub compaction jobs that succeeded.
116 : pub succeeded_sub_compaction_jobs: usize,
117 : /// The layer size before compaction.
118 : pub before_compaction_layer_size: u64,
119 : /// The layer size after compaction.
120 : pub after_compaction_layer_size: u64,
121 : /// The start time of the meta job.
122 : pub start_time: Option<chrono::DateTime<chrono::Utc>>,
123 : /// The end time of the meta job.
124 : pub end_time: Option<chrono::DateTime<chrono::Utc>>,
125 : /// The duration of the meta job.
126 : pub duration_secs: f64,
127 : /// The id of the meta job.
128 : pub meta_job_id: GcCompactionJobId,
129 : /// The LSN below which the layers are compacted, used to compute the statistics.
130 : pub below_lsn: Lsn,
131 : /// The retention ratio of the meta job (after_compaction_layer_size / before_compaction_layer_size)
132 : pub retention_ratio: f64,
133 : }
134 :
135 : impl GcCompactionMetaStatistics {
136 0 : fn finalize(&mut self) {
137 0 : let end_time = chrono::Utc::now();
138 0 : if let Some(start_time) = self.start_time {
139 0 : if end_time > start_time {
140 0 : let delta = end_time - start_time;
141 0 : if let Ok(std_dur) = delta.to_std() {
142 0 : self.duration_secs = std_dur.as_secs_f64();
143 0 : }
144 0 : }
145 0 : }
146 0 : self.retention_ratio = self.after_compaction_layer_size as f64
147 0 : / (self.before_compaction_layer_size as f64 + 1.0);
148 0 : self.end_time = Some(end_time);
149 0 : }
150 : }
151 :
152 : impl GcCompactionQueueItem {
153 0 : pub fn into_compact_info_resp(
154 0 : self,
155 0 : id: GcCompactionJobId,
156 0 : running: bool,
157 0 : ) -> Option<CompactInfoResponse> {
158 0 : match self {
159 0 : GcCompactionQueueItem::MetaJob { options, .. } => Some(CompactInfoResponse {
160 0 : compact_key_range: options.compact_key_range,
161 0 : compact_lsn_range: options.compact_lsn_range,
162 0 : sub_compaction: options.sub_compaction,
163 0 : running,
164 0 : job_id: id.0,
165 0 : }),
166 0 : GcCompactionQueueItem::SubCompactionJob(options) => Some(CompactInfoResponse {
167 0 : compact_key_range: options.compact_key_range,
168 0 : compact_lsn_range: options.compact_lsn_range,
169 0 : sub_compaction: options.sub_compaction,
170 0 : running,
171 0 : job_id: id.0,
172 0 : }),
173 0 : GcCompactionQueueItem::Notify(_, _) => None,
174 : }
175 0 : }
176 : }
177 :
178 : #[derive(Default)]
179 : struct GcCompactionGuardItems {
180 : notify: Option<tokio::sync::oneshot::Sender<()>>,
181 : permit: Option<OwnedSemaphorePermit>,
182 : }
183 :
184 : struct GcCompactionQueueInner {
185 : running: Option<(GcCompactionJobId, GcCompactionQueueItem)>,
186 : queued: VecDeque<(GcCompactionJobId, GcCompactionQueueItem)>,
187 : guards: HashMap<GcCompactionJobId, GcCompactionGuardItems>,
188 : last_id: GcCompactionJobId,
189 : meta_statistics: Option<GcCompactionMetaStatistics>,
190 : }
191 :
192 : impl GcCompactionQueueInner {
193 0 : fn next_id(&mut self) -> GcCompactionJobId {
194 0 : let id = self.last_id;
195 0 : self.last_id = GcCompactionJobId(id.0 + 1);
196 0 : id
197 0 : }
198 : }
199 :
200 : /// A structure to store gc_compaction jobs.
201 : pub struct GcCompactionQueue {
202 : /// All items in the queue, and the currently-running job.
203 : inner: std::sync::Mutex<GcCompactionQueueInner>,
204 : /// Ensure only one thread is consuming the queue.
205 : consumer_lock: tokio::sync::Mutex<()>,
206 : }
207 :
208 0 : static CONCURRENT_GC_COMPACTION_TASKS: Lazy<Arc<Semaphore>> = Lazy::new(|| {
209 0 : // Only allow two timelines on one pageserver to run gc compaction at a time.
210 0 : Arc::new(Semaphore::new(2))
211 0 : });
212 :
213 : impl GcCompactionQueue {
214 0 : pub fn new() -> Self {
215 0 : GcCompactionQueue {
216 0 : inner: std::sync::Mutex::new(GcCompactionQueueInner {
217 0 : running: None,
218 0 : queued: VecDeque::new(),
219 0 : guards: HashMap::new(),
220 0 : last_id: GcCompactionJobId(0),
221 0 : meta_statistics: None,
222 0 : }),
223 0 : consumer_lock: tokio::sync::Mutex::new(()),
224 0 : }
225 0 : }
226 :
227 0 : pub fn cancel_scheduled(&self) {
228 0 : let mut guard = self.inner.lock().unwrap();
229 0 : guard.queued.clear();
230 0 : // TODO: if there is a running job, we should keep the gc guard. However, currently, the cancel
231 0 : // API is only used for testing purposes, so we can drop everything here.
232 0 : guard.guards.clear();
233 0 : }
234 :
235 : /// Schedule a manual compaction job.
236 0 : pub fn schedule_manual_compaction(
237 0 : &self,
238 0 : options: CompactOptions,
239 0 : notify: Option<tokio::sync::oneshot::Sender<()>>,
240 0 : ) -> GcCompactionJobId {
241 0 : let mut guard = self.inner.lock().unwrap();
242 0 : let id = guard.next_id();
243 0 : guard.queued.push_back((
244 0 : id,
245 0 : GcCompactionQueueItem::MetaJob {
246 0 : options,
247 0 : auto: false,
248 0 : },
249 0 : ));
250 0 : guard.guards.entry(id).or_default().notify = notify;
251 0 : info!("scheduled compaction job id={}", id);
252 0 : id
253 0 : }
254 :
255 : /// Schedule an auto compaction job.
256 0 : fn schedule_auto_compaction(
257 0 : &self,
258 0 : options: CompactOptions,
259 0 : permit: OwnedSemaphorePermit,
260 0 : ) -> GcCompactionJobId {
261 0 : let mut guard = self.inner.lock().unwrap();
262 0 : let id = guard.next_id();
263 0 : guard.queued.push_back((
264 0 : id,
265 0 : GcCompactionQueueItem::MetaJob {
266 0 : options,
267 0 : auto: true,
268 0 : },
269 0 : ));
270 0 : guard.guards.entry(id).or_default().permit = Some(permit);
271 0 : id
272 0 : }
273 :
274 : /// Trigger an auto compaction.
275 0 : pub async fn trigger_auto_compaction(
276 0 : &self,
277 0 : timeline: &Arc<Timeline>,
278 0 : ) -> Result<(), CompactionError> {
279 0 : let GcCompactionCombinedSettings {
280 0 : gc_compaction_enabled,
281 0 : gc_compaction_initial_threshold_kb,
282 0 : gc_compaction_ratio_percent,
283 0 : ..
284 0 : } = timeline.get_gc_compaction_settings();
285 0 : if !gc_compaction_enabled {
286 0 : return Ok(());
287 0 : }
288 0 : if self.remaining_jobs_num() > 0 {
289 : // Only schedule auto compaction when the queue is empty
290 0 : return Ok(());
291 0 : }
292 0 : if timeline.ancestor_timeline().is_some() {
293 : // Do not trigger auto compaction for child timelines. We haven't tested
294 : // it enough in staging yet.
295 0 : return Ok(());
296 0 : }
297 0 : if timeline.get_gc_compaction_watermark() == Lsn::INVALID {
298 : // If the gc watermark is not set, we don't need to trigger auto compaction.
299 : // This check is the same as in `gc_compaction_split_jobs` but we don't log
300 : // here and we can also skip the computation of the trigger condition earlier.
301 0 : return Ok(());
302 0 : }
303 :
304 0 : let Ok(permit) = CONCURRENT_GC_COMPACTION_TASKS.clone().try_acquire_owned() else {
305 : // Only allow one compaction run at a time. TODO: As we do `try_acquire_owned`, we cannot ensure
306 : // the fairness of the lock across timelines. We should listen for both `acquire` and `l0_compaction_trigger`
307 : // to ensure the fairness while avoid starving other tasks.
308 0 : return Ok(());
309 : };
310 :
311 0 : let gc_compaction_state = timeline.get_gc_compaction_state();
312 0 : let l2_lsn = gc_compaction_state
313 0 : .map(|x| x.last_completed_lsn)
314 0 : .unwrap_or(Lsn::INVALID);
315 :
316 0 : let layers = {
317 0 : let guard = timeline.layers.read().await;
318 0 : let layer_map = guard.layer_map()?;
319 0 : layer_map.iter_historic_layers().collect_vec()
320 0 : };
321 0 : let mut l2_size: u64 = 0;
322 0 : let mut l1_size = 0;
323 0 : let gc_cutoff = *timeline.get_applied_gc_cutoff_lsn();
324 0 : for layer in layers {
325 0 : if layer.lsn_range.start <= l2_lsn {
326 0 : l2_size += layer.file_size();
327 0 : } else if layer.lsn_range.start <= gc_cutoff {
328 0 : l1_size += layer.file_size();
329 0 : }
330 : }
331 :
332 0 : fn trigger_compaction(
333 0 : l1_size: u64,
334 0 : l2_size: u64,
335 0 : gc_compaction_initial_threshold_kb: u64,
336 0 : gc_compaction_ratio_percent: u64,
337 0 : ) -> bool {
338 : const AUTO_TRIGGER_LIMIT: u64 = 150 * 1024 * 1024 * 1024; // 150GB
339 0 : if l1_size + l2_size >= AUTO_TRIGGER_LIMIT {
340 : // Do not auto-trigger when physical size >= 150GB
341 0 : return false;
342 0 : }
343 0 : // initial trigger
344 0 : if l2_size == 0 && l1_size >= gc_compaction_initial_threshold_kb * 1024 {
345 0 : info!(
346 0 : "trigger auto-compaction because l1_size={} >= gc_compaction_initial_threshold_kb={}",
347 : l1_size, gc_compaction_initial_threshold_kb
348 : );
349 0 : return true;
350 0 : }
351 0 : // size ratio trigger
352 0 : if l2_size == 0 {
353 0 : return false;
354 0 : }
355 0 : if l1_size as f64 / l2_size as f64 >= (gc_compaction_ratio_percent as f64 / 100.0) {
356 0 : info!(
357 0 : "trigger auto-compaction because l1_size={} / l2_size={} > gc_compaction_ratio_percent={}",
358 : l1_size, l2_size, gc_compaction_ratio_percent
359 : );
360 0 : return true;
361 0 : }
362 0 : false
363 0 : }
364 :
365 0 : if trigger_compaction(
366 0 : l1_size,
367 0 : l2_size,
368 0 : gc_compaction_initial_threshold_kb,
369 0 : gc_compaction_ratio_percent,
370 0 : ) {
371 0 : self.schedule_auto_compaction(
372 0 : CompactOptions {
373 0 : flags: {
374 0 : let mut flags = EnumSet::new();
375 0 : flags |= CompactFlags::EnhancedGcBottomMostCompaction;
376 0 : if timeline.get_compaction_l0_first() {
377 0 : flags |= CompactFlags::YieldForL0;
378 0 : }
379 0 : flags
380 0 : },
381 0 : sub_compaction: true,
382 0 : // Only auto-trigger gc-compaction over the data keyspace due to concerns in
383 0 : // https://github.com/neondatabase/neon/issues/11318.
384 0 : compact_key_range: Some(CompactKeyRange {
385 0 : start: Key::MIN,
386 0 : end: Key::metadata_key_range().start,
387 0 : }),
388 0 : compact_lsn_range: None,
389 0 : sub_compaction_max_job_size_mb: None,
390 0 : },
391 0 : permit,
392 0 : );
393 0 : info!(
394 0 : "scheduled auto gc-compaction: l1_size={}, l2_size={}, l2_lsn={}, gc_cutoff={}",
395 : l1_size, l2_size, l2_lsn, gc_cutoff
396 : );
397 : } else {
398 0 : debug!(
399 0 : "did not trigger auto gc-compaction: l1_size={}, l2_size={}, l2_lsn={}, gc_cutoff={}",
400 : l1_size, l2_size, l2_lsn, gc_cutoff
401 : );
402 : }
403 0 : Ok(())
404 0 : }
405 :
406 0 : async fn collect_layer_below_lsn(
407 0 : &self,
408 0 : timeline: &Arc<Timeline>,
409 0 : lsn: Lsn,
410 0 : ) -> Result<u64, CompactionError> {
411 0 : let guard = timeline.layers.read().await;
412 0 : let layer_map = guard.layer_map()?;
413 0 : let layers = layer_map.iter_historic_layers().collect_vec();
414 0 : let mut size = 0;
415 0 : for layer in layers {
416 0 : if layer.lsn_range.start <= lsn {
417 0 : size += layer.file_size();
418 0 : }
419 : }
420 0 : Ok(size)
421 0 : }
422 :
423 : /// Notify the caller the job has finished and unblock GC.
424 0 : fn notify_and_unblock(&self, id: GcCompactionJobId) {
425 0 : info!("compaction job id={} finished", id);
426 0 : let mut guard = self.inner.lock().unwrap();
427 0 : if let Some(items) = guard.guards.remove(&id) {
428 0 : if let Some(tx) = items.notify {
429 0 : let _ = tx.send(());
430 0 : }
431 0 : }
432 0 : if let Some(ref meta_statistics) = guard.meta_statistics {
433 0 : if meta_statistics.meta_job_id == id {
434 0 : if let Ok(stats) = serde_json::to_string(&meta_statistics) {
435 0 : info!(
436 0 : "gc-compaction meta statistics for job id = {}: {}",
437 : id, stats
438 : );
439 0 : }
440 0 : }
441 0 : }
442 0 : }
443 :
444 0 : fn clear_running_job(&self) {
445 0 : let mut guard = self.inner.lock().unwrap();
446 0 : guard.running = None;
447 0 : }
448 :
449 0 : async fn handle_sub_compaction(
450 0 : &self,
451 0 : id: GcCompactionJobId,
452 0 : options: CompactOptions,
453 0 : timeline: &Arc<Timeline>,
454 0 : auto: bool,
455 0 : ) -> Result<(), CompactionError> {
456 0 : info!(
457 0 : "running scheduled enhanced gc bottom-most compaction with sub-compaction, splitting compaction jobs"
458 : );
459 0 : let res = timeline
460 0 : .gc_compaction_split_jobs(
461 0 : GcCompactJob::from_compact_options(options.clone()),
462 0 : options.sub_compaction_max_job_size_mb,
463 0 : )
464 0 : .await;
465 0 : let jobs = match res {
466 0 : Ok(jobs) => jobs,
467 0 : Err(err) => {
468 0 : warn!("cannot split gc-compaction jobs: {}, unblocked gc", err);
469 0 : self.notify_and_unblock(id);
470 0 : return Err(err);
471 : }
472 : };
473 0 : if jobs.is_empty() {
474 0 : info!("no jobs to run, skipping scheduled compaction task");
475 0 : self.notify_and_unblock(id);
476 : } else {
477 0 : let jobs_len = jobs.len();
478 0 : let mut pending_tasks = Vec::new();
479 0 : // gc-compaction might pick more layers or fewer layers to compact. The L2 LSN does not need to be accurate.
480 0 : // And therefore, we simply assume the maximum LSN of all jobs is the expected L2 LSN.
481 0 : let expected_l2_lsn = jobs
482 0 : .iter()
483 0 : .map(|job| job.compact_lsn_range.end)
484 0 : .max()
485 0 : .unwrap();
486 0 : for job in jobs {
487 : // Unfortunately we need to convert the `GcCompactJob` back to `CompactionOptions`
488 : // until we do further refactors to allow directly call `compact_with_gc`.
489 0 : let mut flags: EnumSet<CompactFlags> = EnumSet::default();
490 0 : flags |= CompactFlags::EnhancedGcBottomMostCompaction;
491 0 : if job.dry_run {
492 0 : flags |= CompactFlags::DryRun;
493 0 : }
494 0 : if options.flags.contains(CompactFlags::YieldForL0) {
495 0 : flags |= CompactFlags::YieldForL0;
496 0 : }
497 0 : let options = CompactOptions {
498 0 : flags,
499 0 : sub_compaction: false,
500 0 : compact_key_range: Some(job.compact_key_range.into()),
501 0 : compact_lsn_range: Some(job.compact_lsn_range.into()),
502 0 : sub_compaction_max_job_size_mb: None,
503 0 : };
504 0 : pending_tasks.push(GcCompactionQueueItem::SubCompactionJob(options));
505 : }
506 :
507 0 : if !auto {
508 0 : pending_tasks.push(GcCompactionQueueItem::Notify(id, None));
509 0 : } else {
510 0 : pending_tasks.push(GcCompactionQueueItem::Notify(id, Some(expected_l2_lsn)));
511 0 : }
512 :
513 0 : let layer_size = self
514 0 : .collect_layer_below_lsn(timeline, expected_l2_lsn)
515 0 : .await?;
516 :
517 : {
518 0 : let mut guard = self.inner.lock().unwrap();
519 0 : let mut tasks = Vec::new();
520 0 : for task in pending_tasks {
521 0 : let id = guard.next_id();
522 0 : tasks.push((id, task));
523 0 : }
524 0 : tasks.reverse();
525 0 : for item in tasks {
526 0 : guard.queued.push_front(item);
527 0 : }
528 0 : guard.meta_statistics = Some(GcCompactionMetaStatistics {
529 0 : meta_job_id: id,
530 0 : start_time: Some(chrono::Utc::now()),
531 0 : before_compaction_layer_size: layer_size,
532 0 : below_lsn: expected_l2_lsn,
533 0 : total_sub_compaction_jobs: jobs_len,
534 0 : ..Default::default()
535 0 : });
536 0 : }
537 0 :
538 0 : info!(
539 0 : "scheduled enhanced gc bottom-most compaction with sub-compaction, split into {} jobs",
540 : jobs_len
541 : );
542 : }
543 0 : Ok(())
544 0 : }
545 :
546 : /// Take a job from the queue and process it. Returns if there are still pending tasks.
547 0 : pub async fn iteration(
548 0 : &self,
549 0 : cancel: &CancellationToken,
550 0 : ctx: &RequestContext,
551 0 : gc_block: &GcBlock,
552 0 : timeline: &Arc<Timeline>,
553 0 : ) -> Result<CompactionOutcome, CompactionError> {
554 0 : let res = self.iteration_inner(cancel, ctx, gc_block, timeline).await;
555 0 : if let Err(err) = &res {
556 0 : log_compaction_error(err, None, cancel.is_cancelled(), true);
557 0 : }
558 0 : match res {
559 0 : Ok(res) => Ok(res),
560 0 : Err(CompactionError::ShuttingDown) => Err(CompactionError::ShuttingDown),
561 : Err(_) => {
562 : // There are some cases where traditional gc might collect some layer
563 : // files causing gc-compaction cannot read the full history of the key.
564 : // This needs to be resolved in the long-term by improving the compaction
565 : // process. For now, let's simply avoid such errors triggering the
566 : // circuit breaker.
567 0 : Ok(CompactionOutcome::Skipped)
568 : }
569 : }
570 0 : }
571 :
572 0 : async fn iteration_inner(
573 0 : &self,
574 0 : cancel: &CancellationToken,
575 0 : ctx: &RequestContext,
576 0 : gc_block: &GcBlock,
577 0 : timeline: &Arc<Timeline>,
578 0 : ) -> Result<CompactionOutcome, CompactionError> {
579 0 : let Ok(_one_op_at_a_time_guard) = self.consumer_lock.try_lock() else {
580 0 : return Err(CompactionError::AlreadyRunning(
581 0 : "cannot run gc-compaction because another gc-compaction is running. This should not happen because we only call this function from the gc-compaction queue.",
582 0 : ));
583 : };
584 : let has_pending_tasks;
585 0 : let mut yield_for_l0 = false;
586 0 : let Some((id, item)) = ({
587 0 : let mut guard = self.inner.lock().unwrap();
588 0 : if let Some((id, item)) = guard.queued.pop_front() {
589 0 : guard.running = Some((id, item.clone()));
590 0 : has_pending_tasks = !guard.queued.is_empty();
591 0 : Some((id, item))
592 : } else {
593 0 : has_pending_tasks = false;
594 0 : None
595 : }
596 : }) else {
597 0 : self.trigger_auto_compaction(timeline).await?;
598 : // Always yield after triggering auto-compaction. Gc-compaction is a low-priority task and we
599 : // have not implemented preemption mechanism yet. We always want to yield it to more important
600 : // tasks if there is one.
601 0 : return Ok(CompactionOutcome::Done);
602 : };
603 0 : match item {
604 0 : GcCompactionQueueItem::MetaJob { options, auto } => {
605 0 : if !options
606 0 : .flags
607 0 : .contains(CompactFlags::EnhancedGcBottomMostCompaction)
608 : {
609 0 : warn!(
610 0 : "ignoring scheduled compaction task: scheduled task must be gc compaction: {:?}",
611 : options
612 : );
613 0 : } else if options.sub_compaction {
614 0 : info!(
615 0 : "running scheduled enhanced gc bottom-most compaction with sub-compaction, splitting compaction jobs"
616 : );
617 0 : self.handle_sub_compaction(id, options, timeline, auto)
618 0 : .await?;
619 : } else {
620 : // Auto compaction always enables sub-compaction so we don't need to handle update_l2_lsn
621 : // in this branch.
622 0 : let _gc_guard = match gc_block.start().await {
623 0 : Ok(guard) => guard,
624 0 : Err(e) => {
625 0 : self.notify_and_unblock(id);
626 0 : self.clear_running_job();
627 0 : return Err(CompactionError::Other(anyhow!(
628 0 : "cannot run gc-compaction because gc is blocked: {}",
629 0 : e
630 0 : )));
631 : }
632 : };
633 0 : let res = timeline.compact_with_options(cancel, options, ctx).await;
634 0 : let compaction_result = match res {
635 0 : Ok(res) => res,
636 0 : Err(err) => {
637 0 : warn!(%err, "failed to run gc-compaction");
638 0 : self.notify_and_unblock(id);
639 0 : self.clear_running_job();
640 0 : return Err(err);
641 : }
642 : };
643 0 : if compaction_result == CompactionOutcome::YieldForL0 {
644 0 : yield_for_l0 = true;
645 0 : }
646 : }
647 : }
648 0 : GcCompactionQueueItem::SubCompactionJob(options) => {
649 : // TODO: error handling, clear the queue if any task fails?
650 0 : let _gc_guard = match gc_block.start().await {
651 0 : Ok(guard) => guard,
652 0 : Err(e) => {
653 0 : self.clear_running_job();
654 0 : return Err(CompactionError::Other(anyhow!(
655 0 : "cannot run gc-compaction because gc is blocked: {}",
656 0 : e
657 0 : )));
658 : }
659 : };
660 0 : let res = timeline.compact_with_options(cancel, options, ctx).await;
661 0 : let compaction_result = match res {
662 0 : Ok(res) => res,
663 0 : Err(err) => {
664 0 : warn!(%err, "failed to run gc-compaction subcompaction job");
665 0 : self.clear_running_job();
666 0 : let mut guard = self.inner.lock().unwrap();
667 0 : if let Some(ref mut meta_statistics) = guard.meta_statistics {
668 0 : meta_statistics.failed_sub_compaction_jobs += 1;
669 0 : }
670 0 : return Err(err);
671 : }
672 : };
673 0 : if compaction_result == CompactionOutcome::YieldForL0 {
674 0 : // We will permenantly give up a task if we yield for L0 compaction: the preempted subcompaction job won't be running
675 0 : // again. This ensures that we don't keep doing duplicated work within gc-compaction. Not directly returning here because
676 0 : // we need to clean things up before returning from the function.
677 0 : yield_for_l0 = true;
678 0 : }
679 : {
680 0 : let mut guard = self.inner.lock().unwrap();
681 0 : if let Some(ref mut meta_statistics) = guard.meta_statistics {
682 0 : meta_statistics.succeeded_sub_compaction_jobs += 1;
683 0 : }
684 : }
685 : }
686 0 : GcCompactionQueueItem::Notify(id, l2_lsn) => {
687 0 : let below_lsn = {
688 0 : let mut guard = self.inner.lock().unwrap();
689 0 : if let Some(ref mut meta_statistics) = guard.meta_statistics {
690 0 : meta_statistics.below_lsn
691 : } else {
692 0 : Lsn::INVALID
693 : }
694 : };
695 0 : let layer_size = if below_lsn != Lsn::INVALID {
696 0 : self.collect_layer_below_lsn(timeline, below_lsn).await?
697 : } else {
698 0 : 0
699 : };
700 : {
701 0 : let mut guard = self.inner.lock().unwrap();
702 0 : if let Some(ref mut meta_statistics) = guard.meta_statistics {
703 0 : meta_statistics.after_compaction_layer_size = layer_size;
704 0 : meta_statistics.finalize();
705 0 : }
706 : }
707 0 : self.notify_and_unblock(id);
708 0 : if let Some(l2_lsn) = l2_lsn {
709 0 : let current_l2_lsn = timeline
710 0 : .get_gc_compaction_state()
711 0 : .map(|x| x.last_completed_lsn)
712 0 : .unwrap_or(Lsn::INVALID);
713 0 : if l2_lsn >= current_l2_lsn {
714 0 : info!("l2_lsn updated to {}", l2_lsn);
715 0 : timeline
716 0 : .update_gc_compaction_state(GcCompactionState {
717 0 : last_completed_lsn: l2_lsn,
718 0 : })
719 0 : .map_err(CompactionError::Other)?;
720 : } else {
721 0 : warn!(
722 0 : "l2_lsn updated to {} but it is less than the current l2_lsn {}",
723 : l2_lsn, current_l2_lsn
724 : );
725 : }
726 0 : }
727 : }
728 : }
729 0 : self.clear_running_job();
730 0 : Ok(if yield_for_l0 {
731 0 : tracing::info!("give up gc-compaction: yield for L0 compaction");
732 0 : CompactionOutcome::YieldForL0
733 0 : } else if has_pending_tasks {
734 0 : CompactionOutcome::Pending
735 : } else {
736 0 : CompactionOutcome::Done
737 : })
738 0 : }
739 :
740 : #[allow(clippy::type_complexity)]
741 0 : pub fn remaining_jobs(
742 0 : &self,
743 0 : ) -> (
744 0 : Option<(GcCompactionJobId, GcCompactionQueueItem)>,
745 0 : VecDeque<(GcCompactionJobId, GcCompactionQueueItem)>,
746 0 : ) {
747 0 : let guard = self.inner.lock().unwrap();
748 0 : (guard.running.clone(), guard.queued.clone())
749 0 : }
750 :
751 0 : pub fn remaining_jobs_num(&self) -> usize {
752 0 : let guard = self.inner.lock().unwrap();
753 0 : guard.queued.len() + if guard.running.is_some() { 1 } else { 0 }
754 0 : }
755 : }
756 :
757 : /// A job description for the gc-compaction job. This structure describes the rectangle range that the job will
758 : /// process. The exact layers that need to be compacted/rewritten will be generated when `compact_with_gc` gets
759 : /// called.
760 : #[derive(Debug, Clone)]
761 : pub(crate) struct GcCompactJob {
762 : pub dry_run: bool,
763 : /// The key range to be compacted. The compaction algorithm will only regenerate key-value pairs within this range
764 : /// [left inclusive, right exclusive), and other pairs will be rewritten into new files if necessary.
765 : pub compact_key_range: Range<Key>,
766 : /// The LSN range to be compacted. The compaction algorithm will use this range to determine the layers to be
767 : /// selected for the compaction, and it does not guarantee the generated layers will have exactly the same LSN range
768 : /// as specified here. The true range being compacted is `min_lsn/max_lsn` in [`GcCompactionJobDescription`].
769 : /// min_lsn will always <= the lower bound specified here, and max_lsn will always >= the upper bound specified here.
770 : pub compact_lsn_range: Range<Lsn>,
771 : }
772 :
773 : impl GcCompactJob {
774 336 : pub fn from_compact_options(options: CompactOptions) -> Self {
775 336 : GcCompactJob {
776 336 : dry_run: options.flags.contains(CompactFlags::DryRun),
777 336 : compact_key_range: options
778 336 : .compact_key_range
779 336 : .map(|x| x.into())
780 336 : .unwrap_or(Key::MIN..Key::MAX),
781 336 : compact_lsn_range: options
782 336 : .compact_lsn_range
783 336 : .map(|x| x.into())
784 336 : .unwrap_or(Lsn::INVALID..Lsn::MAX),
785 336 : }
786 336 : }
787 : }
788 :
789 : /// A job description for the gc-compaction job. This structure is generated when `compact_with_gc` is called
790 : /// and contains the exact layers we want to compact.
791 : pub struct GcCompactionJobDescription {
792 : /// All layers to read in the compaction job
793 : selected_layers: Vec<Layer>,
794 : /// GC cutoff of the job. This is the lowest LSN that will be accessed by the read/GC path and we need to
795 : /// keep all deltas <= this LSN or generate an image == this LSN.
796 : gc_cutoff: Lsn,
797 : /// LSNs to retain for the job. Read path will use this LSN so we need to keep deltas <= this LSN or
798 : /// generate an image == this LSN.
799 : retain_lsns_below_horizon: Vec<Lsn>,
800 : /// Maximum layer LSN processed in this compaction, that is max(end_lsn of layers). Exclusive. All data
801 : /// \>= this LSN will be kept and will not be rewritten.
802 : max_layer_lsn: Lsn,
803 : /// Minimum layer LSN processed in this compaction, that is min(start_lsn of layers). Inclusive.
804 : /// All access below (strict lower than `<`) this LSN will be routed through the normal read path instead of
805 : /// k-merge within gc-compaction.
806 : min_layer_lsn: Lsn,
807 : /// Only compact layers overlapping with this range.
808 : compaction_key_range: Range<Key>,
809 : /// When partial compaction is enabled, these layers need to be rewritten to ensure no overlap.
810 : /// This field is here solely for debugging. The field will not be read once the compaction
811 : /// description is generated.
812 : rewrite_layers: Vec<Arc<PersistentLayerDesc>>,
813 : }
814 :
815 : /// The result of bottom-most compaction for a single key at each LSN.
816 : #[derive(Debug)]
817 : #[cfg_attr(test, derive(PartialEq))]
818 : pub struct KeyLogAtLsn(pub Vec<(Lsn, Value)>);
819 :
820 : /// The result of bottom-most compaction.
821 : #[derive(Debug)]
822 : #[cfg_attr(test, derive(PartialEq))]
823 : pub(crate) struct KeyHistoryRetention {
824 : /// Stores logs to reconstruct the value at the given LSN, that is to say, logs <= LSN or image == LSN.
825 : pub(crate) below_horizon: Vec<(Lsn, KeyLogAtLsn)>,
826 : /// Stores logs to reconstruct the value at any LSN above the horizon, that is to say, log > LSN.
827 : pub(crate) above_horizon: KeyLogAtLsn,
828 : }
829 :
830 : impl KeyHistoryRetention {
831 : /// Hack: skip delta layer if we need to produce a layer of a same key-lsn.
832 : ///
833 : /// This can happen if we have removed some deltas in "the middle" of some existing layer's key-lsn-range.
834 : /// For example, consider the case where a single delta with range [0x10,0x50) exists.
835 : /// And we have branches at LSN 0x10, 0x20, 0x30.
836 : /// Then we delete branch @ 0x20.
837 : /// Bottom-most compaction may now delete the delta [0x20,0x30).
838 : /// And that wouldnt' change the shape of the layer.
839 : ///
840 : /// Note that bottom-most-gc-compaction never _adds_ new data in that case, only removes.
841 : ///
842 : /// `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.
843 444 : async fn discard_key(key: &PersistentLayerKey, tline: &Arc<Timeline>, dry_run: bool) -> bool {
844 444 : if dry_run {
845 0 : return true;
846 444 : }
847 444 : if LayerMap::is_l0(&key.key_range, key.is_delta) {
848 : // gc-compaction should not produce L0 deltas, otherwise it will break the layer order.
849 : // We should ignore such layers.
850 0 : return true;
851 444 : }
852 : let layer_generation;
853 : {
854 444 : let guard = tline.layers.read().await;
855 444 : if !guard.contains_key(key) {
856 312 : return false;
857 132 : }
858 132 : layer_generation = guard.get_from_key(key).metadata().generation;
859 132 : }
860 132 : if layer_generation == tline.generation {
861 132 : info!(
862 : key=%key,
863 : ?layer_generation,
864 0 : "discard layer due to duplicated layer key in the same generation",
865 : );
866 132 : true
867 : } else {
868 0 : false
869 : }
870 444 : }
871 :
872 : /// Pipe a history of a single key to the writers.
873 : ///
874 : /// If `image_writer` is none, the images will be placed into the delta layers.
875 : /// The delta writer will contain all images and deltas (below and above the horizon) except the bottom-most images.
876 : #[allow(clippy::too_many_arguments)]
877 3828 : async fn pipe_to(
878 3828 : self,
879 3828 : key: Key,
880 3828 : delta_writer: &mut SplitDeltaLayerWriter<'_>,
881 3828 : mut image_writer: Option<&mut SplitImageLayerWriter<'_>>,
882 3828 : stat: &mut CompactionStatistics,
883 3828 : ctx: &RequestContext,
884 3828 : ) -> anyhow::Result<()> {
885 3828 : let mut first_batch = true;
886 12264 : for (cutoff_lsn, KeyLogAtLsn(logs)) in self.below_horizon {
887 8436 : if first_batch {
888 3828 : if logs.len() == 1 && logs[0].1.is_image() {
889 3600 : let Value::Image(img) = &logs[0].1 else {
890 0 : unreachable!()
891 : };
892 3600 : stat.produce_image_key(img);
893 3600 : if let Some(image_writer) = image_writer.as_mut() {
894 3600 : image_writer.put_image(key, img.clone(), ctx).await?;
895 : } else {
896 0 : delta_writer
897 0 : .put_value(key, cutoff_lsn, Value::Image(img.clone()), ctx)
898 0 : .await?;
899 : }
900 : } else {
901 396 : for (lsn, val) in logs {
902 168 : stat.produce_key(&val);
903 168 : delta_writer.put_value(key, lsn, val, ctx).await?;
904 : }
905 : }
906 3828 : first_batch = false;
907 : } else {
908 5304 : for (lsn, val) in logs {
909 696 : stat.produce_key(&val);
910 696 : delta_writer.put_value(key, lsn, val, ctx).await?;
911 : }
912 : }
913 : }
914 3828 : let KeyLogAtLsn(above_horizon_logs) = self.above_horizon;
915 4176 : for (lsn, val) in above_horizon_logs {
916 348 : stat.produce_key(&val);
917 348 : delta_writer.put_value(key, lsn, val, ctx).await?;
918 : }
919 3828 : Ok(())
920 3828 : }
921 :
922 : /// Verify if every key in the retention is readable by replaying the logs.
923 3876 : async fn verify(
924 3876 : &self,
925 3876 : key: Key,
926 3876 : base_img_from_ancestor: &Option<(Key, Lsn, Bytes)>,
927 3876 : full_history: &[(Key, Lsn, Value)],
928 3876 : tline: &Arc<Timeline>,
929 3876 : ) -> anyhow::Result<()> {
930 : // Usually the min_lsn should be the first record but we do a full iteration to be safe.
931 5496 : let Some(min_lsn) = full_history.iter().map(|(_, lsn, _)| *lsn).min() else {
932 : // This should never happen b/c if we don't have any history of a key, we won't even do `generate_key_retention`.
933 0 : return Ok(());
934 : };
935 5496 : let Some(max_lsn) = full_history.iter().map(|(_, lsn, _)| *lsn).max() else {
936 : // This should never happen b/c if we don't have any history of a key, we won't even do `generate_key_retention`.
937 0 : return Ok(());
938 : };
939 3876 : let mut base_img = base_img_from_ancestor
940 3876 : .as_ref()
941 3876 : .map(|(_, lsn, img)| (*lsn, img));
942 3876 : let mut history = Vec::new();
943 :
944 12324 : async fn collect_and_verify(
945 12324 : key: Key,
946 12324 : lsn: Lsn,
947 12324 : base_img: &Option<(Lsn, &Bytes)>,
948 12324 : history: &[(Lsn, &NeonWalRecord)],
949 12324 : tline: &Arc<Timeline>,
950 12324 : skip_empty: bool,
951 12324 : ) -> anyhow::Result<()> {
952 12324 : if base_img.is_none() && history.is_empty() {
953 0 : if skip_empty {
954 0 : return Ok(());
955 0 : }
956 0 : anyhow::bail!("verification failed: key {} has no history at {}", key, lsn);
957 12324 : };
958 12324 :
959 12324 : let mut records = history
960 12324 : .iter()
961 12324 : .map(|(lsn, val)| (*lsn, (*val).clone()))
962 12324 : .collect::<Vec<_>>();
963 12324 :
964 12324 : // WAL redo requires records in the reverse LSN order
965 12324 : records.reverse();
966 12324 : let data = ValueReconstructState {
967 12324 : img: base_img.as_ref().map(|(lsn, img)| (*lsn, (*img).clone())),
968 12324 : records,
969 12324 : };
970 12324 :
971 12324 : tline
972 12324 : .reconstruct_value(key, lsn, data, RedoAttemptType::GcCompaction)
973 12324 : .await
974 12324 : .with_context(|| format!("verification failed for key {} at lsn {}", key, lsn))?;
975 :
976 12324 : Ok(())
977 12324 : }
978 :
979 12432 : for (retain_lsn, KeyLogAtLsn(logs)) in &self.below_horizon {
980 13152 : for (lsn, val) in logs {
981 912 : match val {
982 3684 : Value::Image(img) => {
983 3684 : base_img = Some((*lsn, img));
984 3684 : history.clear();
985 3684 : }
986 912 : Value::WalRecord(rec) if val.will_init() => {
987 0 : base_img = None;
988 0 : history.clear();
989 0 : history.push((*lsn, rec));
990 0 : }
991 912 : Value::WalRecord(rec) => {
992 912 : history.push((*lsn, rec));
993 912 : }
994 : }
995 : }
996 8556 : if *retain_lsn >= min_lsn {
997 : // Only verify after the key appears in the full history for the first time.
998 :
999 : // We don't modify history: in theory, we could replace the history with a single
1000 : // image as in `generate_key_retention` to make redos at later LSNs faster. But we
1001 : // want to verify everything as if they are read from the real layer map.
1002 8388 : collect_and_verify(key, *retain_lsn, &base_img, &history, tline, false)
1003 8388 : .await
1004 8388 : .context("below horizon retain_lsn")?;
1005 168 : }
1006 : }
1007 :
1008 4320 : for (lsn, val) in &self.above_horizon.0 {
1009 384 : match val {
1010 60 : Value::Image(img) => {
1011 60 : // Above the GC horizon, we verify every time we see an image.
1012 60 : collect_and_verify(key, *lsn, &base_img, &history, tline, true)
1013 60 : .await
1014 60 : .context("above horizon full image")?;
1015 60 : base_img = Some((*lsn, img));
1016 60 : history.clear();
1017 : }
1018 384 : Value::WalRecord(rec) if val.will_init() => {
1019 0 : // Above the GC horizon, we verify every time we see an init record.
1020 0 : collect_and_verify(key, *lsn, &base_img, &history, tline, true)
1021 0 : .await
1022 0 : .context("above horizon init record")?;
1023 0 : base_img = None;
1024 0 : history.clear();
1025 0 : history.push((*lsn, rec));
1026 : }
1027 384 : Value::WalRecord(rec) => {
1028 384 : history.push((*lsn, rec));
1029 384 : }
1030 : }
1031 : }
1032 : // Ensure the latest record is readable.
1033 3876 : collect_and_verify(key, max_lsn, &base_img, &history, tline, false)
1034 3876 : .await
1035 3876 : .context("latest record")?;
1036 3876 : Ok(())
1037 3876 : }
1038 : }
1039 :
1040 : #[derive(Debug, Serialize, Default)]
1041 : struct CompactionStatisticsNumSize {
1042 : num: u64,
1043 : size: u64,
1044 : }
1045 :
1046 : #[derive(Debug, Serialize, Default)]
1047 : pub struct CompactionStatistics {
1048 : /// Delta layer visited (maybe compressed, physical size)
1049 : delta_layer_visited: CompactionStatisticsNumSize,
1050 : /// Image layer visited (maybe compressed, physical size)
1051 : image_layer_visited: CompactionStatisticsNumSize,
1052 : /// Delta layer produced (maybe compressed, physical size)
1053 : delta_layer_produced: CompactionStatisticsNumSize,
1054 : /// Image layer produced (maybe compressed, physical size)
1055 : image_layer_produced: CompactionStatisticsNumSize,
1056 : /// Delta layer discarded (maybe compressed, physical size of the layer being discarded instead of the original layer)
1057 : delta_layer_discarded: CompactionStatisticsNumSize,
1058 : /// Image layer discarded (maybe compressed, physical size of the layer being discarded instead of the original layer)
1059 : image_layer_discarded: CompactionStatisticsNumSize,
1060 : num_unique_keys_visited: usize,
1061 : /// Delta visited (uncompressed, original size)
1062 : wal_keys_visited: CompactionStatisticsNumSize,
1063 : /// Image visited (uncompressed, original size)
1064 : image_keys_visited: CompactionStatisticsNumSize,
1065 : /// Delta produced (uncompressed, original size)
1066 : wal_produced: CompactionStatisticsNumSize,
1067 : /// Image produced (uncompressed, original size)
1068 : image_produced: CompactionStatisticsNumSize,
1069 :
1070 : // Time spent in each phase
1071 : time_acquire_lock_secs: f64,
1072 : time_analyze_secs: f64,
1073 : time_download_layer_secs: f64,
1074 : time_to_first_kv_pair_secs: f64,
1075 : time_main_loop_secs: f64,
1076 : time_final_phase_secs: f64,
1077 : time_total_secs: f64,
1078 :
1079 : // Summary
1080 : /// Ratio of the key-value size after/before gc-compaction.
1081 : uncompressed_retention_ratio: f64,
1082 : /// Ratio of the physical size after/before gc-compaction.
1083 : compressed_retention_ratio: f64,
1084 : }
1085 :
1086 : impl CompactionStatistics {
1087 6408 : fn estimated_size_of_value(val: &Value) -> usize {
1088 2628 : match val {
1089 3780 : Value::Image(img) => img.len(),
1090 0 : Value::WalRecord(NeonWalRecord::Postgres { rec, .. }) => rec.len(),
1091 2628 : _ => std::mem::size_of::<NeonWalRecord>(),
1092 : }
1093 6408 : }
1094 10068 : fn estimated_size_of_key() -> usize {
1095 10068 : KEY_SIZE // TODO: distinguish image layer and delta layer (count LSN in delta layer)
1096 10068 : }
1097 528 : fn visit_delta_layer(&mut self, size: u64) {
1098 528 : self.delta_layer_visited.num += 1;
1099 528 : self.delta_layer_visited.size += size;
1100 528 : }
1101 420 : fn visit_image_layer(&mut self, size: u64) {
1102 420 : self.image_layer_visited.num += 1;
1103 420 : self.image_layer_visited.size += size;
1104 420 : }
1105 3840 : fn on_unique_key_visited(&mut self) {
1106 3840 : self.num_unique_keys_visited += 1;
1107 3840 : }
1108 1476 : fn visit_wal_key(&mut self, val: &Value) {
1109 1476 : self.wal_keys_visited.num += 1;
1110 1476 : self.wal_keys_visited.size +=
1111 1476 : Self::estimated_size_of_value(val) as u64 + Self::estimated_size_of_key() as u64;
1112 1476 : }
1113 3780 : fn visit_image_key(&mut self, val: &Value) {
1114 3780 : self.image_keys_visited.num += 1;
1115 3780 : self.image_keys_visited.size +=
1116 3780 : Self::estimated_size_of_value(val) as u64 + Self::estimated_size_of_key() as u64;
1117 3780 : }
1118 1212 : fn produce_key(&mut self, val: &Value) {
1119 1212 : match val {
1120 60 : Value::Image(img) => self.produce_image_key(img),
1121 1152 : Value::WalRecord(_) => self.produce_wal_key(val),
1122 : }
1123 1212 : }
1124 1152 : fn produce_wal_key(&mut self, val: &Value) {
1125 1152 : self.wal_produced.num += 1;
1126 1152 : self.wal_produced.size +=
1127 1152 : Self::estimated_size_of_value(val) as u64 + Self::estimated_size_of_key() as u64;
1128 1152 : }
1129 3660 : fn produce_image_key(&mut self, val: &Bytes) {
1130 3660 : self.image_produced.num += 1;
1131 3660 : self.image_produced.size += val.len() as u64 + Self::estimated_size_of_key() as u64;
1132 3660 : }
1133 84 : fn discard_delta_layer(&mut self, original_size: u64) {
1134 84 : self.delta_layer_discarded.num += 1;
1135 84 : self.delta_layer_discarded.size += original_size;
1136 84 : }
1137 48 : fn discard_image_layer(&mut self, original_size: u64) {
1138 48 : self.image_layer_discarded.num += 1;
1139 48 : self.image_layer_discarded.size += original_size;
1140 48 : }
1141 144 : fn produce_delta_layer(&mut self, size: u64) {
1142 144 : self.delta_layer_produced.num += 1;
1143 144 : self.delta_layer_produced.size += size;
1144 144 : }
1145 180 : fn produce_image_layer(&mut self, size: u64) {
1146 180 : self.image_layer_produced.num += 1;
1147 180 : self.image_layer_produced.size += size;
1148 180 : }
1149 312 : fn finalize(&mut self) {
1150 312 : let original_key_value_size = self.image_keys_visited.size + self.wal_keys_visited.size;
1151 312 : let produced_key_value_size = self.image_produced.size + self.wal_produced.size;
1152 312 : self.uncompressed_retention_ratio =
1153 312 : produced_key_value_size as f64 / (original_key_value_size as f64 + 1.0); // avoid div by 0
1154 312 : let original_physical_size = self.image_layer_visited.size + self.delta_layer_visited.size;
1155 312 : let produced_physical_size = self.image_layer_produced.size
1156 312 : + self.delta_layer_produced.size
1157 312 : + self.image_layer_discarded.size
1158 312 : + self.delta_layer_discarded.size; // Also include the discarded layers to make the ratio accurate
1159 312 : self.compressed_retention_ratio =
1160 312 : produced_physical_size as f64 / (original_physical_size as f64 + 1.0); // avoid div by 0
1161 312 : }
1162 : }
1163 :
1164 : #[derive(Default, Debug, Clone, Copy, PartialEq, Eq)]
1165 : pub enum CompactionOutcome {
1166 : #[default]
1167 : /// No layers need to be compacted after this round. Compaction doesn't need
1168 : /// to be immediately scheduled.
1169 : Done,
1170 : /// Still has pending layers to be compacted after this round. Ideally, the scheduler
1171 : /// should immediately schedule another compaction.
1172 : Pending,
1173 : /// A timeline needs L0 compaction. Yield and schedule an immediate L0 compaction pass (only
1174 : /// guaranteed when `compaction_l0_first` is enabled).
1175 : YieldForL0,
1176 : /// Compaction was skipped, because the timeline is ineligible for compaction.
1177 : Skipped,
1178 : }
1179 :
1180 : impl Timeline {
1181 : /// TODO: cancellation
1182 : ///
1183 : /// Returns whether the compaction has pending tasks.
1184 2184 : pub(crate) async fn compact_legacy(
1185 2184 : self: &Arc<Self>,
1186 2184 : cancel: &CancellationToken,
1187 2184 : options: CompactOptions,
1188 2184 : ctx: &RequestContext,
1189 2184 : ) -> Result<CompactionOutcome, CompactionError> {
1190 2184 : if options
1191 2184 : .flags
1192 2184 : .contains(CompactFlags::EnhancedGcBottomMostCompaction)
1193 : {
1194 0 : self.compact_with_gc(cancel, options, ctx).await?;
1195 0 : return Ok(CompactionOutcome::Done);
1196 2184 : }
1197 2184 :
1198 2184 : if options.flags.contains(CompactFlags::DryRun) {
1199 0 : return Err(CompactionError::Other(anyhow!(
1200 0 : "dry-run mode is not supported for legacy compaction for now"
1201 0 : )));
1202 2184 : }
1203 2184 :
1204 2184 : if options.compact_key_range.is_some() || options.compact_lsn_range.is_some() {
1205 : // maybe useful in the future? could implement this at some point
1206 0 : return Err(CompactionError::Other(anyhow!(
1207 0 : "compaction range is not supported for legacy compaction for now"
1208 0 : )));
1209 2184 : }
1210 2184 :
1211 2184 : // High level strategy for compaction / image creation:
1212 2184 : //
1213 2184 : // 1. First, do a L0 compaction to ensure we move the L0
1214 2184 : // layers into the historic layer map get flat levels of
1215 2184 : // layers. If we did not compact all L0 layers, we will
1216 2184 : // prioritize compacting the timeline again and not do
1217 2184 : // any of the compactions below.
1218 2184 : //
1219 2184 : // 2. Then, calculate the desired "partitioning" of the
1220 2184 : // currently in-use key space. The goal is to partition the
1221 2184 : // key space into roughly fixed-size chunks, but also take into
1222 2184 : // account any existing image layers, and try to align the
1223 2184 : // chunk boundaries with the existing image layers to avoid
1224 2184 : // too much churn. Also try to align chunk boundaries with
1225 2184 : // relation boundaries. In principle, we don't know about
1226 2184 : // relation boundaries here, we just deal with key-value
1227 2184 : // pairs, and the code in pgdatadir_mapping.rs knows how to
1228 2184 : // map relations into key-value pairs. But in practice we know
1229 2184 : // that 'field6' is the block number, and the fields 1-5
1230 2184 : // identify a relation. This is just an optimization,
1231 2184 : // though.
1232 2184 : //
1233 2184 : // 3. Once we know the partitioning, for each partition,
1234 2184 : // decide if it's time to create a new image layer. The
1235 2184 : // criteria is: there has been too much "churn" since the last
1236 2184 : // image layer? The "churn" is fuzzy concept, it's a
1237 2184 : // combination of too many delta files, or too much WAL in
1238 2184 : // total in the delta file. Or perhaps: if creating an image
1239 2184 : // file would allow to delete some older files.
1240 2184 : //
1241 2184 : // 4. In the end, if the tenant gets auto-sharded, we will run
1242 2184 : // a shard-ancestor compaction.
1243 2184 :
1244 2184 : // Is the timeline being deleted?
1245 2184 : if self.is_stopping() {
1246 0 : trace!("Dropping out of compaction on timeline shutdown");
1247 0 : return Err(CompactionError::ShuttingDown);
1248 2184 : }
1249 2184 :
1250 2184 : let target_file_size = self.get_checkpoint_distance();
1251 :
1252 : // Define partitioning schema if needed
1253 :
1254 : // 1. L0 Compact
1255 2184 : let l0_outcome = {
1256 2184 : let timer = self.metrics.compact_time_histo.start_timer();
1257 2184 : let l0_outcome = self
1258 2184 : .compact_level0(
1259 2184 : target_file_size,
1260 2184 : options.flags.contains(CompactFlags::ForceL0Compaction),
1261 2184 : ctx,
1262 2184 : )
1263 2184 : .await?;
1264 2184 : timer.stop_and_record();
1265 2184 : l0_outcome
1266 2184 : };
1267 2184 :
1268 2184 : if options.flags.contains(CompactFlags::OnlyL0Compaction) {
1269 0 : return Ok(l0_outcome);
1270 2184 : }
1271 2184 :
1272 2184 : // Yield if we have pending L0 compaction. The scheduler will do another pass.
1273 2184 : if (l0_outcome == CompactionOutcome::Pending || l0_outcome == CompactionOutcome::YieldForL0)
1274 0 : && options.flags.contains(CompactFlags::YieldForL0)
1275 : {
1276 0 : info!("image/ancestor compaction yielding for L0 compaction");
1277 0 : return Ok(CompactionOutcome::YieldForL0);
1278 2184 : }
1279 2184 :
1280 2184 : // 2. Repartition and create image layers if necessary
1281 2184 : match self
1282 2184 : .repartition(
1283 2184 : self.get_last_record_lsn(),
1284 2184 : self.get_compaction_target_size(),
1285 2184 : options.flags,
1286 2184 : ctx,
1287 2184 : )
1288 2184 : .await
1289 : {
1290 2184 : Ok(((dense_partitioning, sparse_partitioning), lsn)) => {
1291 2184 : // Disables access_stats updates, so that the files we read remain candidates for eviction after we're done with them
1292 2184 : let image_ctx = RequestContextBuilder::from(ctx)
1293 2184 : .access_stats_behavior(AccessStatsBehavior::Skip)
1294 2184 : .attached_child();
1295 2184 :
1296 2184 : let mut partitioning = dense_partitioning;
1297 2184 : partitioning
1298 2184 : .parts
1299 2184 : .extend(sparse_partitioning.into_dense().parts);
1300 :
1301 : // 3. Create new image layers for partitions that have been modified "enough".
1302 2184 : let (image_layers, outcome) = self
1303 2184 : .create_image_layers(
1304 2184 : &partitioning,
1305 2184 : lsn,
1306 2184 : if options
1307 2184 : .flags
1308 2184 : .contains(CompactFlags::ForceImageLayerCreation)
1309 : {
1310 84 : ImageLayerCreationMode::Force
1311 : } else {
1312 2100 : ImageLayerCreationMode::Try
1313 : },
1314 2184 : &image_ctx,
1315 2184 : self.last_image_layer_creation_status
1316 2184 : .load()
1317 2184 : .as_ref()
1318 2184 : .clone(),
1319 2184 : options.flags.contains(CompactFlags::YieldForL0),
1320 2184 : )
1321 2184 : .await
1322 2184 : .inspect_err(|err| {
1323 : if let CreateImageLayersError::GetVectoredError(
1324 : GetVectoredError::MissingKey(_),
1325 0 : ) = err
1326 : {
1327 0 : critical!("missing key during compaction: {err:?}");
1328 0 : }
1329 2184 : })?;
1330 :
1331 2184 : self.last_image_layer_creation_status
1332 2184 : .store(Arc::new(outcome.clone()));
1333 2184 :
1334 2184 : self.upload_new_image_layers(image_layers)?;
1335 2184 : if let LastImageLayerCreationStatus::Incomplete { .. } = outcome {
1336 : // Yield and do not do any other kind of compaction.
1337 0 : info!(
1338 0 : "skipping shard ancestor compaction due to pending image layer generation tasks (preempted by L0 compaction)."
1339 : );
1340 0 : return Ok(CompactionOutcome::YieldForL0);
1341 2184 : }
1342 : }
1343 :
1344 : // Suppress errors when cancelled.
1345 0 : Err(_) if self.cancel.is_cancelled() => {}
1346 0 : Err(err) if err.is_cancel() => {}
1347 :
1348 : // Alert on critical errors that indicate data corruption.
1349 0 : Err(err) if err.is_critical() => {
1350 0 : critical!("could not compact, repartitioning keyspace failed: {err:?}");
1351 : }
1352 :
1353 : // Log other errors. No partitioning? This is normal, if the timeline was just created
1354 : // as an empty timeline. Also in unit tests, when we use the timeline as a simple
1355 : // key-value store, ignoring the datadir layout. Log the error but continue.
1356 0 : Err(err) => error!("could not compact, repartitioning keyspace failed: {err:?}"),
1357 : };
1358 :
1359 2184 : let partition_count = self.partitioning.read().0.0.parts.len();
1360 2184 :
1361 2184 : // 4. Shard ancestor compaction
1362 2184 : if self.get_compaction_shard_ancestor() && self.shard_identity.count >= ShardCount::new(2) {
1363 : // Limit the number of layer rewrites to the number of partitions: this means its
1364 : // runtime should be comparable to a full round of image layer creations, rather than
1365 : // being potentially much longer.
1366 0 : let rewrite_max = partition_count;
1367 :
1368 0 : let outcome = self
1369 0 : .compact_shard_ancestors(
1370 0 : rewrite_max,
1371 0 : options.flags.contains(CompactFlags::YieldForL0),
1372 0 : ctx,
1373 0 : )
1374 0 : .await?;
1375 0 : match outcome {
1376 0 : CompactionOutcome::Pending | CompactionOutcome::YieldForL0 => return Ok(outcome),
1377 0 : CompactionOutcome::Done | CompactionOutcome::Skipped => {}
1378 : }
1379 2184 : }
1380 :
1381 2184 : Ok(CompactionOutcome::Done)
1382 2184 : }
1383 :
1384 : /// Check for layers that are elegible to be rewritten:
1385 : /// - Shard splitting: After a shard split, ancestor layers beyond pitr_interval, so that
1386 : /// we don't indefinitely retain keys in this shard that aren't needed.
1387 : /// - For future use: layers beyond pitr_interval that are in formats we would
1388 : /// rather not maintain compatibility with indefinitely.
1389 : ///
1390 : /// Note: this phase may read and write many gigabytes of data: use rewrite_max to bound
1391 : /// how much work it will try to do in each compaction pass.
1392 0 : async fn compact_shard_ancestors(
1393 0 : self: &Arc<Self>,
1394 0 : rewrite_max: usize,
1395 0 : yield_for_l0: bool,
1396 0 : ctx: &RequestContext,
1397 0 : ) -> Result<CompactionOutcome, CompactionError> {
1398 0 : let mut outcome = CompactionOutcome::Done;
1399 0 : let mut drop_layers = Vec::new();
1400 0 : let mut layers_to_rewrite: Vec<Layer> = Vec::new();
1401 0 :
1402 0 : // We will use the Lsn cutoff of the last GC as a threshold for rewriting layers: if a
1403 0 : // layer is behind this Lsn, it indicates that the layer is being retained beyond the
1404 0 : // pitr_interval, for example because a branchpoint references it.
1405 0 : //
1406 0 : // Holding this read guard also blocks [`Self::gc_timeline`] from entering while we
1407 0 : // are rewriting layers.
1408 0 : let latest_gc_cutoff = self.get_applied_gc_cutoff_lsn();
1409 0 : let pitr_cutoff = self.gc_info.read().unwrap().cutoffs.time;
1410 :
1411 0 : let layers = self.layers.read().await;
1412 0 : let layers_iter = layers.layer_map()?.iter_historic_layers();
1413 0 : let (layers_total, mut layers_checked) = (layers_iter.len(), 0);
1414 0 : for layer_desc in layers_iter {
1415 0 : layers_checked += 1;
1416 0 : let layer = layers.get_from_desc(&layer_desc);
1417 0 : if layer.metadata().shard.shard_count == self.shard_identity.count {
1418 : // This layer does not belong to a historic ancestor, no need to re-image it.
1419 0 : continue;
1420 0 : }
1421 0 :
1422 0 : // This layer was created on an ancestor shard: check if it contains any data for this shard.
1423 0 : let sharded_range = ShardedRange::new(layer_desc.get_key_range(), &self.shard_identity);
1424 0 : let layer_local_page_count = sharded_range.page_count();
1425 0 : let layer_raw_page_count = ShardedRange::raw_size(&layer_desc.get_key_range());
1426 0 : if layer_local_page_count == 0 {
1427 : // This ancestral layer only covers keys that belong to other shards.
1428 : // We include the full metadata in the log: if we had some critical bug that caused
1429 : // us to incorrectly drop layers, this would simplify manually debugging + reinstating those layers.
1430 0 : debug!(%layer, old_metadata=?layer.metadata(),
1431 0 : "dropping layer after shard split, contains no keys for this shard",
1432 : );
1433 :
1434 0 : if cfg!(debug_assertions) {
1435 : // Expensive, exhaustive check of keys in this layer: this guards against ShardedRange's calculations being
1436 : // wrong. If ShardedRange claims the local page count is zero, then no keys in this layer
1437 : // should be !is_key_disposable()
1438 : // TODO: exclude sparse keyspace from this check, otherwise it will infinitely loop.
1439 0 : let range = layer_desc.get_key_range();
1440 0 : let mut key = range.start;
1441 0 : while key < range.end {
1442 0 : debug_assert!(self.shard_identity.is_key_disposable(&key));
1443 0 : key = key.next();
1444 : }
1445 0 : }
1446 :
1447 0 : drop_layers.push(layer);
1448 0 : continue;
1449 0 : } else if layer_local_page_count != u32::MAX
1450 0 : && layer_local_page_count == layer_raw_page_count
1451 : {
1452 0 : debug!(%layer,
1453 0 : "layer is entirely shard local ({} keys), no need to filter it",
1454 : layer_local_page_count
1455 : );
1456 0 : continue;
1457 0 : }
1458 0 :
1459 0 : // Only rewrite a layer if we can reclaim significant space.
1460 0 : if layer_local_page_count != u32::MAX
1461 0 : && layer_local_page_count as f64 / layer_raw_page_count as f64
1462 0 : <= ANCESTOR_COMPACTION_REWRITE_THRESHOLD
1463 : {
1464 0 : debug!(%layer,
1465 0 : "layer has a large share of local pages \
1466 0 : ({layer_local_page_count}/{layer_raw_page_count} > \
1467 0 : {ANCESTOR_COMPACTION_REWRITE_THRESHOLD}), not rewriting",
1468 : );
1469 0 : }
1470 :
1471 : // Don't bother re-writing a layer if it is within the PITR window: it will age-out eventually
1472 : // without incurring the I/O cost of a rewrite.
1473 0 : if layer_desc.get_lsn_range().end >= *latest_gc_cutoff {
1474 0 : debug!(%layer, "Skipping rewrite of layer still in GC window ({} >= {})",
1475 0 : layer_desc.get_lsn_range().end, *latest_gc_cutoff);
1476 0 : continue;
1477 0 : }
1478 0 :
1479 0 : // We do not yet implement rewrite of delta layers.
1480 0 : if layer_desc.is_delta() {
1481 0 : debug!(%layer, "Skipping rewrite of delta layer");
1482 0 : continue;
1483 0 : }
1484 0 :
1485 0 : // We don't bother rewriting layers that aren't visible, since these won't be needed by
1486 0 : // reads and will likely be garbage collected soon.
1487 0 : if layer.visibility() != LayerVisibilityHint::Visible {
1488 0 : debug!(%layer, "Skipping rewrite of invisible layer");
1489 0 : continue;
1490 0 : }
1491 0 :
1492 0 : // Only rewrite layers if their generations differ. This guarantees:
1493 0 : // - that local rewrite is safe, as local layer paths will differ between existing layer and rewritten one
1494 0 : // - that the layer is persistent in remote storage, as we only see old-generation'd layer via loading from remote storage
1495 0 : if layer.metadata().generation == self.generation {
1496 0 : debug!(%layer, "Skipping rewrite, is not from old generation");
1497 0 : continue;
1498 0 : }
1499 0 :
1500 0 : if layers_to_rewrite.len() >= rewrite_max {
1501 0 : debug!(%layer, "Will rewrite layer on a future compaction, already rewrote {}",
1502 0 : layers_to_rewrite.len()
1503 : );
1504 0 : outcome = CompactionOutcome::Pending;
1505 0 : break;
1506 0 : }
1507 0 :
1508 0 : // Fall through: all our conditions for doing a rewrite passed.
1509 0 : layers_to_rewrite.push(layer);
1510 : }
1511 :
1512 : // Drop read lock on layer map before we start doing time-consuming I/O.
1513 0 : drop(layers);
1514 0 :
1515 0 : // Drop out early if there's nothing to do.
1516 0 : if layers_to_rewrite.is_empty() && drop_layers.is_empty() {
1517 0 : return Ok(CompactionOutcome::Done);
1518 0 : }
1519 0 :
1520 0 : info!(
1521 0 : "starting shard ancestor compaction, rewriting {} layers and dropping {} layers, \
1522 0 : checked {layers_checked}/{layers_total} layers \
1523 0 : (latest_gc_cutoff={} pitr_cutoff={})",
1524 0 : layers_to_rewrite.len(),
1525 0 : drop_layers.len(),
1526 0 : *latest_gc_cutoff,
1527 : pitr_cutoff,
1528 : );
1529 0 : let started = Instant::now();
1530 0 :
1531 0 : let mut replace_image_layers = Vec::new();
1532 :
1533 0 : for layer in layers_to_rewrite {
1534 0 : if self.cancel.is_cancelled() {
1535 0 : return Err(CompactionError::ShuttingDown);
1536 0 : }
1537 0 :
1538 0 : info!(layer=%layer, "rewriting layer after shard split");
1539 0 : let mut image_layer_writer = ImageLayerWriter::new(
1540 0 : self.conf,
1541 0 : self.timeline_id,
1542 0 : self.tenant_shard_id,
1543 0 : &layer.layer_desc().key_range,
1544 0 : layer.layer_desc().image_layer_lsn(),
1545 0 : &self.gate,
1546 0 : self.cancel.clone(),
1547 0 : ctx,
1548 0 : )
1549 0 : .await
1550 0 : .map_err(CompactionError::Other)?;
1551 :
1552 : // Safety of layer rewrites:
1553 : // - We are writing to a different local file path than we are reading from, so the old Layer
1554 : // cannot interfere with the new one.
1555 : // - In the page cache, contents for a particular VirtualFile are stored with a file_id that
1556 : // is different for two layers with the same name (in `ImageLayerInner::new` we always
1557 : // acquire a fresh id from [`crate::page_cache::next_file_id`]. So readers do not risk
1558 : // reading the index from one layer file, and then data blocks from the rewritten layer file.
1559 : // - Any readers that have a reference to the old layer will keep it alive until they are done
1560 : // with it. If they are trying to promote from remote storage, that will fail, but this is the same
1561 : // as for compaction generally: compaction is allowed to delete layers that readers might be trying to use.
1562 : // - We do not run concurrently with other kinds of compaction, so the only layer map writes we race with are:
1563 : // - GC, which at worst witnesses us "undelete" a layer that they just deleted.
1564 : // - ingestion, which only inserts layers, therefore cannot collide with us.
1565 0 : let resident = layer.download_and_keep_resident(ctx).await?;
1566 :
1567 0 : let keys_written = resident
1568 0 : .filter(&self.shard_identity, &mut image_layer_writer, ctx)
1569 0 : .await?;
1570 :
1571 0 : if keys_written > 0 {
1572 0 : let (desc, path) = image_layer_writer
1573 0 : .finish(ctx)
1574 0 : .await
1575 0 : .map_err(CompactionError::Other)?;
1576 0 : let new_layer = Layer::finish_creating(self.conf, self, desc, &path)
1577 0 : .map_err(CompactionError::Other)?;
1578 0 : info!(layer=%new_layer, "rewrote layer, {} -> {} bytes",
1579 0 : layer.metadata().file_size,
1580 0 : new_layer.metadata().file_size);
1581 :
1582 0 : replace_image_layers.push((layer, new_layer));
1583 0 : } else {
1584 0 : // Drop the old layer. Usually for this case we would already have noticed that
1585 0 : // the layer has no data for us with the ShardedRange check above, but
1586 0 : drop_layers.push(layer);
1587 0 : }
1588 :
1589 : // Yield for L0 compaction if necessary, but make sure we update the layer map below
1590 : // with the work we've already done.
1591 0 : if yield_for_l0
1592 0 : && self
1593 0 : .l0_compaction_trigger
1594 0 : .notified()
1595 0 : .now_or_never()
1596 0 : .is_some()
1597 : {
1598 0 : info!("shard ancestor compaction yielding for L0 compaction");
1599 0 : outcome = CompactionOutcome::YieldForL0;
1600 0 : break;
1601 0 : }
1602 : }
1603 :
1604 0 : for layer in &drop_layers {
1605 0 : info!(%layer, old_metadata=?layer.metadata(),
1606 0 : "dropping layer after shard split (no keys for this shard)",
1607 : );
1608 : }
1609 :
1610 : // At this point, we have replaced local layer files with their rewritten form, but not yet uploaded
1611 : // metadata to reflect that. If we restart here, the replaced layer files will look invalid (size mismatch
1612 : // to remote index) and be removed. This is inefficient but safe.
1613 0 : fail::fail_point!("compact-shard-ancestors-localonly");
1614 0 :
1615 0 : // Update the LayerMap so that readers will use the new layers, and enqueue it for writing to remote storage
1616 0 : self.rewrite_layers(replace_image_layers, drop_layers)
1617 0 : .await?;
1618 :
1619 0 : fail::fail_point!("compact-shard-ancestors-enqueued");
1620 0 :
1621 0 : // We wait for all uploads to complete before finishing this compaction stage. This is not
1622 0 : // necessary for correctness, but it simplifies testing, and avoids proceeding with another
1623 0 : // Timeline's compaction while this timeline's uploads may be generating lots of disk I/O
1624 0 : // load.
1625 0 : if outcome != CompactionOutcome::YieldForL0 {
1626 0 : info!("shard ancestor compaction waiting for uploads");
1627 0 : tokio::select! {
1628 0 : result = self.remote_client.wait_completion() => match result {
1629 0 : Ok(()) => {},
1630 0 : Err(WaitCompletionError::NotInitialized(ni)) => return Err(CompactionError::from(ni)),
1631 : Err(WaitCompletionError::UploadQueueShutDownOrStopped) => {
1632 0 : return Err(CompactionError::ShuttingDown);
1633 : }
1634 : },
1635 : // Don't wait if there's L0 compaction to do. We don't need to update the outcome
1636 : // here, because we've already done the actual work.
1637 0 : _ = self.l0_compaction_trigger.notified(), if yield_for_l0 => {},
1638 : }
1639 0 : }
1640 :
1641 0 : info!(
1642 0 : "shard ancestor compaction done in {:.3}s{}",
1643 0 : started.elapsed().as_secs_f64(),
1644 0 : match outcome {
1645 : CompactionOutcome::Pending =>
1646 0 : format!(", with pending work (rewrite_max={rewrite_max})"),
1647 0 : CompactionOutcome::YieldForL0 => String::from(", yielding for L0 compaction"),
1648 0 : CompactionOutcome::Skipped | CompactionOutcome::Done => String::new(),
1649 : }
1650 : );
1651 :
1652 0 : fail::fail_point!("compact-shard-ancestors-persistent");
1653 0 :
1654 0 : Ok(outcome)
1655 0 : }
1656 :
1657 : /// Update the LayerVisibilityHint of layers covered by image layers, based on whether there is
1658 : /// an image layer between them and the most recent readable LSN (branch point or tip of timeline). The
1659 : /// purpose of the visibility hint is to record which layers need to be available to service reads.
1660 : ///
1661 : /// The result may be used as an input to eviction and secondary downloads to de-prioritize layers
1662 : /// that we know won't be needed for reads.
1663 1452 : pub(crate) async fn update_layer_visibility(
1664 1452 : &self,
1665 1452 : ) -> Result<(), super::layer_manager::Shutdown> {
1666 1452 : let head_lsn = self.get_last_record_lsn();
1667 :
1668 : // We will sweep through layers in reverse-LSN order. We only do historic layers. L0 deltas
1669 : // are implicitly left visible, because LayerVisibilityHint's default is Visible, and we never modify it here.
1670 : // Note that L0 deltas _can_ be covered by image layers, but we consider them 'visible' because we anticipate that
1671 : // they will be subject to L0->L1 compaction in the near future.
1672 1452 : let layer_manager = self.layers.read().await;
1673 1452 : let layer_map = layer_manager.layer_map()?;
1674 :
1675 1452 : let readable_points = {
1676 1452 : let children = self.gc_info.read().unwrap().retain_lsns.clone();
1677 1452 :
1678 1452 : let mut readable_points = Vec::with_capacity(children.len() + 1);
1679 1452 : for (child_lsn, _child_timeline_id, is_offloaded) in &children {
1680 0 : if *is_offloaded == MaybeOffloaded::Yes {
1681 0 : continue;
1682 0 : }
1683 0 : readable_points.push(*child_lsn);
1684 : }
1685 1452 : readable_points.push(head_lsn);
1686 1452 : readable_points
1687 1452 : };
1688 1452 :
1689 1452 : let (layer_visibility, covered) = layer_map.get_visibility(readable_points);
1690 3648 : for (layer_desc, visibility) in layer_visibility {
1691 2196 : // FIXME: a more efficiency bulk zip() through the layers rather than NlogN getting each one
1692 2196 : let layer = layer_manager.get_from_desc(&layer_desc);
1693 2196 : layer.set_visibility(visibility);
1694 2196 : }
1695 :
1696 : // TODO: publish our covered KeySpace to our parent, so that when they update their visibility, they can
1697 : // avoid assuming that everything at a branch point is visible.
1698 1452 : drop(covered);
1699 1452 : Ok(())
1700 1452 : }
1701 :
1702 : /// Collect a bunch of Level 0 layer files, and compact and reshuffle them as
1703 : /// as Level 1 files. Returns whether the L0 layers are fully compacted.
1704 2184 : async fn compact_level0(
1705 2184 : self: &Arc<Self>,
1706 2184 : target_file_size: u64,
1707 2184 : force_compaction_ignore_threshold: bool,
1708 2184 : ctx: &RequestContext,
1709 2184 : ) -> Result<CompactionOutcome, CompactionError> {
1710 : let CompactLevel0Phase1Result {
1711 2184 : new_layers,
1712 2184 : deltas_to_compact,
1713 2184 : outcome,
1714 : } = {
1715 2184 : let phase1_span = info_span!("compact_level0_phase1");
1716 2184 : let ctx = ctx.attached_child();
1717 2184 : let mut stats = CompactLevel0Phase1StatsBuilder {
1718 2184 : version: Some(2),
1719 2184 : tenant_id: Some(self.tenant_shard_id),
1720 2184 : timeline_id: Some(self.timeline_id),
1721 2184 : ..Default::default()
1722 2184 : };
1723 2184 :
1724 2184 : let begin = tokio::time::Instant::now();
1725 2184 : let phase1_layers_locked = self.layers.read().await;
1726 2184 : let now = tokio::time::Instant::now();
1727 2184 : stats.read_lock_acquisition_micros =
1728 2184 : DurationRecorder::Recorded(RecordedDuration(now - begin), now);
1729 2184 : self.compact_level0_phase1(
1730 2184 : phase1_layers_locked,
1731 2184 : stats,
1732 2184 : target_file_size,
1733 2184 : force_compaction_ignore_threshold,
1734 2184 : &ctx,
1735 2184 : )
1736 2184 : .instrument(phase1_span)
1737 2184 : .await?
1738 : };
1739 :
1740 2184 : if new_layers.is_empty() && deltas_to_compact.is_empty() {
1741 : // nothing to do
1742 2016 : return Ok(CompactionOutcome::Done);
1743 168 : }
1744 168 :
1745 168 : self.finish_compact_batch(&new_layers, &Vec::new(), &deltas_to_compact)
1746 168 : .await?;
1747 168 : Ok(outcome)
1748 2184 : }
1749 :
1750 : /// Level0 files first phase of compaction, explained in the [`Self::compact_legacy`] comment.
1751 2184 : async fn compact_level0_phase1<'a>(
1752 2184 : self: &'a Arc<Self>,
1753 2184 : guard: tokio::sync::RwLockReadGuard<'a, LayerManager>,
1754 2184 : mut stats: CompactLevel0Phase1StatsBuilder,
1755 2184 : target_file_size: u64,
1756 2184 : force_compaction_ignore_threshold: bool,
1757 2184 : ctx: &RequestContext,
1758 2184 : ) -> Result<CompactLevel0Phase1Result, CompactionError> {
1759 2184 : stats.read_lock_held_spawn_blocking_startup_micros =
1760 2184 : stats.read_lock_acquisition_micros.till_now(); // set by caller
1761 2184 : let layers = guard.layer_map()?;
1762 2184 : let level0_deltas = layers.level0_deltas();
1763 2184 : stats.level0_deltas_count = Some(level0_deltas.len());
1764 2184 :
1765 2184 : // Only compact if enough layers have accumulated.
1766 2184 : let threshold = self.get_compaction_threshold();
1767 2184 : if level0_deltas.is_empty() || level0_deltas.len() < threshold {
1768 2016 : if force_compaction_ignore_threshold {
1769 0 : if !level0_deltas.is_empty() {
1770 0 : info!(
1771 0 : level0_deltas = level0_deltas.len(),
1772 0 : threshold, "too few deltas to compact, but forcing compaction"
1773 : );
1774 : } else {
1775 0 : info!(
1776 0 : level0_deltas = level0_deltas.len(),
1777 0 : threshold, "too few deltas to compact, cannot force compaction"
1778 : );
1779 0 : return Ok(CompactLevel0Phase1Result::default());
1780 : }
1781 : } else {
1782 2016 : debug!(
1783 0 : level0_deltas = level0_deltas.len(),
1784 0 : threshold, "too few deltas to compact"
1785 : );
1786 2016 : return Ok(CompactLevel0Phase1Result::default());
1787 : }
1788 168 : }
1789 :
1790 168 : let mut level0_deltas = level0_deltas
1791 168 : .iter()
1792 2412 : .map(|x| guard.get_from_desc(x))
1793 168 : .collect::<Vec<_>>();
1794 168 :
1795 168 : // Gather the files to compact in this iteration.
1796 168 : //
1797 168 : // Start with the oldest Level 0 delta file, and collect any other
1798 168 : // level 0 files that form a contiguous sequence, such that the end
1799 168 : // LSN of previous file matches the start LSN of the next file.
1800 168 : //
1801 168 : // Note that if the files don't form such a sequence, we might
1802 168 : // "compact" just a single file. That's a bit pointless, but it allows
1803 168 : // us to get rid of the level 0 file, and compact the other files on
1804 168 : // the next iteration. This could probably made smarter, but such
1805 168 : // "gaps" in the sequence of level 0 files should only happen in case
1806 168 : // of a crash, partial download from cloud storage, or something like
1807 168 : // that, so it's not a big deal in practice.
1808 4488 : level0_deltas.sort_by_key(|l| l.layer_desc().lsn_range.start);
1809 168 : let mut level0_deltas_iter = level0_deltas.iter();
1810 168 :
1811 168 : let first_level0_delta = level0_deltas_iter.next().unwrap();
1812 168 : let mut prev_lsn_end = first_level0_delta.layer_desc().lsn_range.end;
1813 168 : let mut deltas_to_compact = Vec::with_capacity(level0_deltas.len());
1814 168 :
1815 168 : // Accumulate the size of layers in `deltas_to_compact`
1816 168 : let mut deltas_to_compact_bytes = 0;
1817 168 :
1818 168 : // Under normal circumstances, we will accumulate up to compaction_upper_limit L0s of size
1819 168 : // checkpoint_distance each. To avoid edge cases using extra system resources, bound our
1820 168 : // work in this function to only operate on this much delta data at once.
1821 168 : //
1822 168 : // In general, compaction_threshold should be <= compaction_upper_limit, but in case that
1823 168 : // the constraint is not respected, we use the larger of the two.
1824 168 : let delta_size_limit = std::cmp::max(
1825 168 : self.get_compaction_upper_limit(),
1826 168 : self.get_compaction_threshold(),
1827 168 : ) as u64
1828 168 : * std::cmp::max(self.get_checkpoint_distance(), DEFAULT_CHECKPOINT_DISTANCE);
1829 168 :
1830 168 : let mut fully_compacted = true;
1831 168 :
1832 168 : deltas_to_compact.push(first_level0_delta.download_and_keep_resident(ctx).await?);
1833 2412 : for l in level0_deltas_iter {
1834 2244 : let lsn_range = &l.layer_desc().lsn_range;
1835 2244 :
1836 2244 : if lsn_range.start != prev_lsn_end {
1837 0 : break;
1838 2244 : }
1839 2244 : deltas_to_compact.push(l.download_and_keep_resident(ctx).await?);
1840 2244 : deltas_to_compact_bytes += l.metadata().file_size;
1841 2244 : prev_lsn_end = lsn_range.end;
1842 2244 :
1843 2244 : if deltas_to_compact_bytes >= delta_size_limit {
1844 0 : info!(
1845 0 : l0_deltas_selected = deltas_to_compact.len(),
1846 0 : l0_deltas_total = level0_deltas.len(),
1847 0 : "L0 compaction picker hit max delta layer size limit: {}",
1848 : delta_size_limit
1849 : );
1850 0 : fully_compacted = false;
1851 0 :
1852 0 : // Proceed with compaction, but only a subset of L0s
1853 0 : break;
1854 2244 : }
1855 : }
1856 168 : let lsn_range = Range {
1857 168 : start: deltas_to_compact
1858 168 : .first()
1859 168 : .unwrap()
1860 168 : .layer_desc()
1861 168 : .lsn_range
1862 168 : .start,
1863 168 : end: deltas_to_compact.last().unwrap().layer_desc().lsn_range.end,
1864 168 : };
1865 168 :
1866 168 : info!(
1867 0 : "Starting Level0 compaction in LSN range {}-{} for {} layers ({} deltas in total)",
1868 0 : lsn_range.start,
1869 0 : lsn_range.end,
1870 0 : deltas_to_compact.len(),
1871 0 : level0_deltas.len()
1872 : );
1873 :
1874 2412 : for l in deltas_to_compact.iter() {
1875 2412 : info!("compact includes {l}");
1876 : }
1877 :
1878 : // We don't need the original list of layers anymore. Drop it so that
1879 : // we don't accidentally use it later in the function.
1880 168 : drop(level0_deltas);
1881 168 :
1882 168 : stats.read_lock_held_prerequisites_micros = stats
1883 168 : .read_lock_held_spawn_blocking_startup_micros
1884 168 : .till_now();
1885 :
1886 : // TODO: replace with streaming k-merge
1887 168 : let all_keys = {
1888 168 : let mut all_keys = Vec::new();
1889 2412 : for l in deltas_to_compact.iter() {
1890 2412 : if self.cancel.is_cancelled() {
1891 0 : return Err(CompactionError::ShuttingDown);
1892 2412 : }
1893 2412 : let delta = l.get_as_delta(ctx).await.map_err(CompactionError::Other)?;
1894 2412 : let keys = delta
1895 2412 : .index_entries(ctx)
1896 2412 : .await
1897 2412 : .map_err(CompactionError::Other)?;
1898 2412 : all_keys.extend(keys);
1899 : }
1900 : // The current stdlib sorting implementation is designed in a way where it is
1901 : // particularly fast where the slice is made up of sorted sub-ranges.
1902 26542800 : all_keys.sort_by_key(|DeltaEntry { key, lsn, .. }| (*key, *lsn));
1903 168 : all_keys
1904 168 : };
1905 168 :
1906 168 : stats.read_lock_held_key_sort_micros = stats.read_lock_held_prerequisites_micros.till_now();
1907 :
1908 : // Determine N largest holes where N is number of compacted layers. The vec is sorted by key range start.
1909 : //
1910 : // A hole is a key range for which this compaction doesn't have any WAL records.
1911 : // Our goal in this compaction iteration is to avoid creating L1s that, in terms of their key range,
1912 : // cover the hole, but actually don't contain any WAL records for that key range.
1913 : // The reason is that the mere stack of L1s (`count_deltas`) triggers image layer creation (`create_image_layers`).
1914 : // That image layer creation would be useless for a hole range covered by L1s that don't contain any WAL records.
1915 : //
1916 : // The algorithm chooses holes as follows.
1917 : // - Slide a 2-window over the keys in key orde to get the hole range (=distance between two keys).
1918 : // - Filter: min threshold on range length
1919 : // - Rank: by coverage size (=number of image layers required to reconstruct each key in the range for which we have any data)
1920 : //
1921 : // For more details, intuition, and some ASCII art see https://github.com/neondatabase/neon/pull/3597#discussion_r1112704451
1922 : #[derive(PartialEq, Eq)]
1923 : struct Hole {
1924 : key_range: Range<Key>,
1925 : coverage_size: usize,
1926 : }
1927 168 : let holes: Vec<Hole> = {
1928 : use std::cmp::Ordering;
1929 : impl Ord for Hole {
1930 0 : fn cmp(&self, other: &Self) -> Ordering {
1931 0 : self.coverage_size.cmp(&other.coverage_size).reverse()
1932 0 : }
1933 : }
1934 : impl PartialOrd for Hole {
1935 0 : fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
1936 0 : Some(self.cmp(other))
1937 0 : }
1938 : }
1939 168 : let max_holes = deltas_to_compact.len();
1940 168 : let last_record_lsn = self.get_last_record_lsn();
1941 168 : let min_hole_range = (target_file_size / page_cache::PAGE_SZ as u64) as i128;
1942 168 : let min_hole_coverage_size = 3; // TODO: something more flexible?
1943 168 : // min-heap (reserve space for one more element added before eviction)
1944 168 : let mut heap: BinaryHeap<Hole> = BinaryHeap::with_capacity(max_holes + 1);
1945 168 : let mut prev: Option<Key> = None;
1946 :
1947 12384228 : for &DeltaEntry { key: next_key, .. } in all_keys.iter() {
1948 12384228 : if let Some(prev_key) = prev {
1949 : // just first fast filter, do not create hole entries for metadata keys. The last hole in the
1950 : // compaction is the gap between data key and metadata keys.
1951 12384060 : if next_key.to_i128() - prev_key.to_i128() >= min_hole_range
1952 0 : && !Key::is_metadata_key(&prev_key)
1953 : {
1954 0 : let key_range = prev_key..next_key;
1955 0 : // Measuring hole by just subtraction of i128 representation of key range boundaries
1956 0 : // has not so much sense, because largest holes will corresponds field1/field2 changes.
1957 0 : // But we are mostly interested to eliminate holes which cause generation of excessive image layers.
1958 0 : // That is why it is better to measure size of hole as number of covering image layers.
1959 0 : let coverage_size =
1960 0 : layers.image_coverage(&key_range, last_record_lsn).len();
1961 0 : if coverage_size >= min_hole_coverage_size {
1962 0 : heap.push(Hole {
1963 0 : key_range,
1964 0 : coverage_size,
1965 0 : });
1966 0 : if heap.len() > max_holes {
1967 0 : heap.pop(); // remove smallest hole
1968 0 : }
1969 0 : }
1970 12384060 : }
1971 168 : }
1972 12384228 : prev = Some(next_key.next());
1973 : }
1974 168 : let mut holes = heap.into_vec();
1975 168 : holes.sort_unstable_by_key(|hole| hole.key_range.start);
1976 168 : holes
1977 168 : };
1978 168 : stats.read_lock_held_compute_holes_micros = stats.read_lock_held_key_sort_micros.till_now();
1979 168 : drop_rlock(guard);
1980 168 :
1981 168 : if self.cancel.is_cancelled() {
1982 0 : return Err(CompactionError::ShuttingDown);
1983 168 : }
1984 168 :
1985 168 : stats.read_lock_drop_micros = stats.read_lock_held_compute_holes_micros.till_now();
1986 :
1987 : // This iterator walks through all key-value pairs from all the layers
1988 : // we're compacting, in key, LSN order.
1989 : // If there's both a Value::Image and Value::WalRecord for the same (key,lsn),
1990 : // then the Value::Image is ordered before Value::WalRecord.
1991 168 : let mut all_values_iter = {
1992 168 : let mut deltas = Vec::with_capacity(deltas_to_compact.len());
1993 2412 : for l in deltas_to_compact.iter() {
1994 2412 : let l = l.get_as_delta(ctx).await.map_err(CompactionError::Other)?;
1995 2412 : deltas.push(l);
1996 : }
1997 168 : MergeIterator::create(&deltas, &[], ctx)
1998 168 : };
1999 168 :
2000 168 : // This iterator walks through all keys and is needed to calculate size used by each key
2001 168 : let mut all_keys_iter = all_keys
2002 168 : .iter()
2003 12384228 : .map(|DeltaEntry { key, lsn, size, .. }| (*key, *lsn, *size))
2004 12384060 : .coalesce(|mut prev, cur| {
2005 12384060 : // Coalesce keys that belong to the same key pair.
2006 12384060 : // This ensures that compaction doesn't put them
2007 12384060 : // into different layer files.
2008 12384060 : // Still limit this by the target file size,
2009 12384060 : // so that we keep the size of the files in
2010 12384060 : // check.
2011 12384060 : if prev.0 == cur.0 && prev.2 < target_file_size {
2012 240228 : prev.2 += cur.2;
2013 240228 : Ok(prev)
2014 : } else {
2015 12143832 : Err((prev, cur))
2016 : }
2017 12384060 : });
2018 168 :
2019 168 : // Merge the contents of all the input delta layers into a new set
2020 168 : // of delta layers, based on the current partitioning.
2021 168 : //
2022 168 : // 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.
2023 168 : // It's possible that there is a single key with so many page versions that storing all of them in a single layer file
2024 168 : // would be too large. In that case, we also split on the LSN dimension.
2025 168 : //
2026 168 : // LSN
2027 168 : // ^
2028 168 : // |
2029 168 : // | +-----------+ +--+--+--+--+
2030 168 : // | | | | | | | |
2031 168 : // | +-----------+ | | | | |
2032 168 : // | | | | | | | |
2033 168 : // | +-----------+ ==> | | | | |
2034 168 : // | | | | | | | |
2035 168 : // | +-----------+ | | | | |
2036 168 : // | | | | | | | |
2037 168 : // | +-----------+ +--+--+--+--+
2038 168 : // |
2039 168 : // +--------------> key
2040 168 : //
2041 168 : //
2042 168 : // If one key (X) has a lot of page versions:
2043 168 : //
2044 168 : // LSN
2045 168 : // ^
2046 168 : // | (X)
2047 168 : // | +-----------+ +--+--+--+--+
2048 168 : // | | | | | | | |
2049 168 : // | +-----------+ | | +--+ |
2050 168 : // | | | | | | | |
2051 168 : // | +-----------+ ==> | | | | |
2052 168 : // | | | | | +--+ |
2053 168 : // | +-----------+ | | | | |
2054 168 : // | | | | | | | |
2055 168 : // | +-----------+ +--+--+--+--+
2056 168 : // |
2057 168 : // +--------------> key
2058 168 : // TODO: this actually divides the layers into fixed-size chunks, not
2059 168 : // based on the partitioning.
2060 168 : //
2061 168 : // TODO: we should also opportunistically materialize and
2062 168 : // garbage collect what we can.
2063 168 : let mut new_layers = Vec::new();
2064 168 : let mut prev_key: Option<Key> = None;
2065 168 : let mut writer: Option<DeltaLayerWriter> = None;
2066 168 : let mut key_values_total_size = 0u64;
2067 168 : let mut dup_start_lsn: Lsn = Lsn::INVALID; // start LSN of layer containing values of the single key
2068 168 : let mut dup_end_lsn: Lsn = Lsn::INVALID; // end LSN of layer containing values of the single key
2069 168 : let mut next_hole = 0; // index of next hole in holes vector
2070 168 :
2071 168 : let mut keys = 0;
2072 :
2073 12384396 : while let Some((key, lsn, value)) = all_values_iter
2074 12384396 : .next()
2075 12384396 : .await
2076 12384396 : .map_err(CompactionError::Other)?
2077 : {
2078 12384228 : keys += 1;
2079 12384228 :
2080 12384228 : if keys % 32_768 == 0 && self.cancel.is_cancelled() {
2081 : // avoid hitting the cancellation token on every key. in benches, we end up
2082 : // shuffling an order of million keys per layer, this means we'll check it
2083 : // around tens of times per layer.
2084 0 : return Err(CompactionError::ShuttingDown);
2085 12384228 : }
2086 12384228 :
2087 12384228 : let same_key = prev_key == Some(key);
2088 12384228 : // We need to check key boundaries once we reach next key or end of layer with the same key
2089 12384228 : if !same_key || lsn == dup_end_lsn {
2090 12144000 : let mut next_key_size = 0u64;
2091 12144000 : let is_dup_layer = dup_end_lsn.is_valid();
2092 12144000 : dup_start_lsn = Lsn::INVALID;
2093 12144000 : if !same_key {
2094 12144000 : dup_end_lsn = Lsn::INVALID;
2095 12144000 : }
2096 : // Determine size occupied by this key. We stop at next key or when size becomes larger than target_file_size
2097 12144000 : for (next_key, next_lsn, next_size) in all_keys_iter.by_ref() {
2098 12144000 : next_key_size = next_size;
2099 12144000 : if key != next_key {
2100 12143832 : if dup_end_lsn.is_valid() {
2101 0 : // We are writting segment with duplicates:
2102 0 : // place all remaining values of this key in separate segment
2103 0 : dup_start_lsn = dup_end_lsn; // new segments starts where old stops
2104 0 : dup_end_lsn = lsn_range.end; // there are no more values of this key till end of LSN range
2105 12143832 : }
2106 12143832 : break;
2107 168 : }
2108 168 : key_values_total_size += next_size;
2109 168 : // Check if it is time to split segment: if total keys size is larger than target file size.
2110 168 : // We need to avoid generation of empty segments if next_size > target_file_size.
2111 168 : if key_values_total_size > target_file_size && lsn != next_lsn {
2112 : // Split key between multiple layers: such layer can contain only single key
2113 0 : dup_start_lsn = if dup_end_lsn.is_valid() {
2114 0 : dup_end_lsn // new segment with duplicates starts where old one stops
2115 : } else {
2116 0 : lsn // start with the first LSN for this key
2117 : };
2118 0 : dup_end_lsn = next_lsn; // upper LSN boundary is exclusive
2119 0 : break;
2120 168 : }
2121 : }
2122 : // handle case when loop reaches last key: in this case dup_end is non-zero but dup_start is not set.
2123 12144000 : if dup_end_lsn.is_valid() && !dup_start_lsn.is_valid() {
2124 0 : dup_start_lsn = dup_end_lsn;
2125 0 : dup_end_lsn = lsn_range.end;
2126 12144000 : }
2127 12144000 : if writer.is_some() {
2128 12143832 : let written_size = writer.as_mut().unwrap().size();
2129 12143832 : let contains_hole =
2130 12143832 : next_hole < holes.len() && key >= holes[next_hole].key_range.end;
2131 : // check if key cause layer overflow or contains hole...
2132 12143832 : if is_dup_layer
2133 12143832 : || dup_end_lsn.is_valid()
2134 12143832 : || written_size + key_values_total_size > target_file_size
2135 12142152 : || contains_hole
2136 : {
2137 : // ... if so, flush previous layer and prepare to write new one
2138 1680 : let (desc, path) = writer
2139 1680 : .take()
2140 1680 : .unwrap()
2141 1680 : .finish(prev_key.unwrap().next(), ctx)
2142 1680 : .await
2143 1680 : .map_err(CompactionError::Other)?;
2144 1680 : let new_delta = Layer::finish_creating(self.conf, self, desc, &path)
2145 1680 : .map_err(CompactionError::Other)?;
2146 :
2147 1680 : new_layers.push(new_delta);
2148 1680 : writer = None;
2149 1680 :
2150 1680 : if contains_hole {
2151 0 : // skip hole
2152 0 : next_hole += 1;
2153 1680 : }
2154 12142152 : }
2155 168 : }
2156 : // Remember size of key value because at next iteration we will access next item
2157 12144000 : key_values_total_size = next_key_size;
2158 240228 : }
2159 12384228 : fail_point!("delta-layer-writer-fail-before-finish", |_| {
2160 0 : Err(CompactionError::Other(anyhow::anyhow!(
2161 0 : "failpoint delta-layer-writer-fail-before-finish"
2162 0 : )))
2163 12384228 : });
2164 :
2165 12384228 : if !self.shard_identity.is_key_disposable(&key) {
2166 12384228 : if writer.is_none() {
2167 1848 : if self.cancel.is_cancelled() {
2168 : // to be somewhat responsive to cancellation, check for each new layer
2169 0 : return Err(CompactionError::ShuttingDown);
2170 1848 : }
2171 : // Create writer if not initiaized yet
2172 1848 : writer = Some(
2173 : DeltaLayerWriter::new(
2174 1848 : self.conf,
2175 1848 : self.timeline_id,
2176 1848 : self.tenant_shard_id,
2177 1848 : key,
2178 1848 : if dup_end_lsn.is_valid() {
2179 : // this is a layer containing slice of values of the same key
2180 0 : debug!("Create new dup layer {}..{}", dup_start_lsn, dup_end_lsn);
2181 0 : dup_start_lsn..dup_end_lsn
2182 : } else {
2183 1848 : debug!("Create new layer {}..{}", lsn_range.start, lsn_range.end);
2184 1848 : lsn_range.clone()
2185 : },
2186 1848 : &self.gate,
2187 1848 : self.cancel.clone(),
2188 1848 : ctx,
2189 1848 : )
2190 1848 : .await
2191 1848 : .map_err(CompactionError::Other)?,
2192 : );
2193 :
2194 1848 : keys = 0;
2195 12382380 : }
2196 :
2197 12384228 : writer
2198 12384228 : .as_mut()
2199 12384228 : .unwrap()
2200 12384228 : .put_value(key, lsn, value, ctx)
2201 12384228 : .await
2202 12384228 : .map_err(CompactionError::Other)?;
2203 : } else {
2204 0 : let owner = self.shard_identity.get_shard_number(&key);
2205 0 :
2206 0 : // This happens after a shard split, when we're compacting an L0 created by our parent shard
2207 0 : debug!("dropping key {key} during compaction (it belongs on shard {owner})");
2208 : }
2209 :
2210 12384228 : if !new_layers.is_empty() {
2211 118716 : fail_point!("after-timeline-compacted-first-L1");
2212 12265512 : }
2213 :
2214 12384228 : prev_key = Some(key);
2215 : }
2216 168 : if let Some(writer) = writer {
2217 168 : let (desc, path) = writer
2218 168 : .finish(prev_key.unwrap().next(), ctx)
2219 168 : .await
2220 168 : .map_err(CompactionError::Other)?;
2221 168 : let new_delta = Layer::finish_creating(self.conf, self, desc, &path)
2222 168 : .map_err(CompactionError::Other)?;
2223 168 : new_layers.push(new_delta);
2224 0 : }
2225 :
2226 : // Sync layers
2227 168 : if !new_layers.is_empty() {
2228 : // Print a warning if the created layer is larger than double the target size
2229 : // Add two pages for potential overhead. This should in theory be already
2230 : // accounted for in the target calculation, but for very small targets,
2231 : // we still might easily hit the limit otherwise.
2232 168 : let warn_limit = target_file_size * 2 + page_cache::PAGE_SZ as u64 * 2;
2233 1848 : for layer in new_layers.iter() {
2234 1848 : if layer.layer_desc().file_size > warn_limit {
2235 0 : warn!(
2236 : %layer,
2237 0 : "created delta file of size {} larger than double of target of {target_file_size}", layer.layer_desc().file_size
2238 : );
2239 1848 : }
2240 : }
2241 :
2242 : // The writer.finish() above already did the fsync of the inodes.
2243 : // We just need to fsync the directory in which these inodes are linked,
2244 : // which we know to be the timeline directory.
2245 : //
2246 : // We use fatal_err() below because the after writer.finish() returns with success,
2247 : // the in-memory state of the filesystem already has the layer file in its final place,
2248 : // and subsequent pageserver code could think it's durable while it really isn't.
2249 168 : let timeline_dir = VirtualFile::open(
2250 168 : &self
2251 168 : .conf
2252 168 : .timeline_path(&self.tenant_shard_id, &self.timeline_id),
2253 168 : ctx,
2254 168 : )
2255 168 : .await
2256 168 : .fatal_err("VirtualFile::open for timeline dir fsync");
2257 168 : timeline_dir
2258 168 : .sync_all()
2259 168 : .await
2260 168 : .fatal_err("VirtualFile::sync_all timeline dir");
2261 0 : }
2262 :
2263 168 : stats.write_layer_files_micros = stats.read_lock_drop_micros.till_now();
2264 168 : stats.new_deltas_count = Some(new_layers.len());
2265 1848 : stats.new_deltas_size = Some(new_layers.iter().map(|l| l.layer_desc().file_size).sum());
2266 168 :
2267 168 : match TryInto::<CompactLevel0Phase1Stats>::try_into(stats)
2268 168 : .and_then(|stats| serde_json::to_string(&stats).context("serde_json::to_string"))
2269 : {
2270 168 : Ok(stats_json) => {
2271 168 : info!(
2272 0 : stats_json = stats_json.as_str(),
2273 0 : "compact_level0_phase1 stats available"
2274 : )
2275 : }
2276 0 : Err(e) => {
2277 0 : warn!("compact_level0_phase1 stats failed to serialize: {:#}", e);
2278 : }
2279 : }
2280 :
2281 : // Without this, rustc complains about deltas_to_compact still
2282 : // being borrowed when we `.into_iter()` below.
2283 168 : drop(all_values_iter);
2284 168 :
2285 168 : Ok(CompactLevel0Phase1Result {
2286 168 : new_layers,
2287 168 : deltas_to_compact: deltas_to_compact
2288 168 : .into_iter()
2289 2412 : .map(|x| x.drop_eviction_guard())
2290 168 : .collect::<Vec<_>>(),
2291 168 : outcome: if fully_compacted {
2292 168 : CompactionOutcome::Done
2293 : } else {
2294 0 : CompactionOutcome::Pending
2295 : },
2296 : })
2297 2184 : }
2298 : }
2299 :
2300 : #[derive(Default)]
2301 : struct CompactLevel0Phase1Result {
2302 : new_layers: Vec<ResidentLayer>,
2303 : deltas_to_compact: Vec<Layer>,
2304 : // Whether we have included all L0 layers, or selected only part of them due to the
2305 : // L0 compaction size limit.
2306 : outcome: CompactionOutcome,
2307 : }
2308 :
2309 : #[derive(Default)]
2310 : struct CompactLevel0Phase1StatsBuilder {
2311 : version: Option<u64>,
2312 : tenant_id: Option<TenantShardId>,
2313 : timeline_id: Option<TimelineId>,
2314 : read_lock_acquisition_micros: DurationRecorder,
2315 : read_lock_held_spawn_blocking_startup_micros: DurationRecorder,
2316 : read_lock_held_key_sort_micros: DurationRecorder,
2317 : read_lock_held_prerequisites_micros: DurationRecorder,
2318 : read_lock_held_compute_holes_micros: DurationRecorder,
2319 : read_lock_drop_micros: DurationRecorder,
2320 : write_layer_files_micros: DurationRecorder,
2321 : level0_deltas_count: Option<usize>,
2322 : new_deltas_count: Option<usize>,
2323 : new_deltas_size: Option<u64>,
2324 : }
2325 :
2326 : #[derive(serde::Serialize)]
2327 : struct CompactLevel0Phase1Stats {
2328 : version: u64,
2329 : tenant_id: TenantShardId,
2330 : timeline_id: TimelineId,
2331 : read_lock_acquisition_micros: RecordedDuration,
2332 : read_lock_held_spawn_blocking_startup_micros: RecordedDuration,
2333 : read_lock_held_key_sort_micros: RecordedDuration,
2334 : read_lock_held_prerequisites_micros: RecordedDuration,
2335 : read_lock_held_compute_holes_micros: RecordedDuration,
2336 : read_lock_drop_micros: RecordedDuration,
2337 : write_layer_files_micros: RecordedDuration,
2338 : level0_deltas_count: usize,
2339 : new_deltas_count: usize,
2340 : new_deltas_size: u64,
2341 : }
2342 :
2343 : impl TryFrom<CompactLevel0Phase1StatsBuilder> for CompactLevel0Phase1Stats {
2344 : type Error = anyhow::Error;
2345 :
2346 168 : fn try_from(value: CompactLevel0Phase1StatsBuilder) -> Result<Self, Self::Error> {
2347 168 : Ok(Self {
2348 168 : version: value.version.ok_or_else(|| anyhow!("version not set"))?,
2349 168 : tenant_id: value
2350 168 : .tenant_id
2351 168 : .ok_or_else(|| anyhow!("tenant_id not set"))?,
2352 168 : timeline_id: value
2353 168 : .timeline_id
2354 168 : .ok_or_else(|| anyhow!("timeline_id not set"))?,
2355 168 : read_lock_acquisition_micros: value
2356 168 : .read_lock_acquisition_micros
2357 168 : .into_recorded()
2358 168 : .ok_or_else(|| anyhow!("read_lock_acquisition_micros not set"))?,
2359 168 : read_lock_held_spawn_blocking_startup_micros: value
2360 168 : .read_lock_held_spawn_blocking_startup_micros
2361 168 : .into_recorded()
2362 168 : .ok_or_else(|| anyhow!("read_lock_held_spawn_blocking_startup_micros not set"))?,
2363 168 : read_lock_held_key_sort_micros: value
2364 168 : .read_lock_held_key_sort_micros
2365 168 : .into_recorded()
2366 168 : .ok_or_else(|| anyhow!("read_lock_held_key_sort_micros not set"))?,
2367 168 : read_lock_held_prerequisites_micros: value
2368 168 : .read_lock_held_prerequisites_micros
2369 168 : .into_recorded()
2370 168 : .ok_or_else(|| anyhow!("read_lock_held_prerequisites_micros not set"))?,
2371 168 : read_lock_held_compute_holes_micros: value
2372 168 : .read_lock_held_compute_holes_micros
2373 168 : .into_recorded()
2374 168 : .ok_or_else(|| anyhow!("read_lock_held_compute_holes_micros not set"))?,
2375 168 : read_lock_drop_micros: value
2376 168 : .read_lock_drop_micros
2377 168 : .into_recorded()
2378 168 : .ok_or_else(|| anyhow!("read_lock_drop_micros not set"))?,
2379 168 : write_layer_files_micros: value
2380 168 : .write_layer_files_micros
2381 168 : .into_recorded()
2382 168 : .ok_or_else(|| anyhow!("write_layer_files_micros not set"))?,
2383 168 : level0_deltas_count: value
2384 168 : .level0_deltas_count
2385 168 : .ok_or_else(|| anyhow!("level0_deltas_count not set"))?,
2386 168 : new_deltas_count: value
2387 168 : .new_deltas_count
2388 168 : .ok_or_else(|| anyhow!("new_deltas_count not set"))?,
2389 168 : new_deltas_size: value
2390 168 : .new_deltas_size
2391 168 : .ok_or_else(|| anyhow!("new_deltas_size not set"))?,
2392 : })
2393 168 : }
2394 : }
2395 :
2396 : impl Timeline {
2397 : /// Entry point for new tiered compaction algorithm.
2398 : ///
2399 : /// All the real work is in the implementation in the pageserver_compaction
2400 : /// crate. The code here would apply to any algorithm implemented by the
2401 : /// same interface, but tiered is the only one at the moment.
2402 : ///
2403 : /// TODO: cancellation
2404 0 : pub(crate) async fn compact_tiered(
2405 0 : self: &Arc<Self>,
2406 0 : _cancel: &CancellationToken,
2407 0 : ctx: &RequestContext,
2408 0 : ) -> Result<(), CompactionError> {
2409 0 : let fanout = self.get_compaction_threshold() as u64;
2410 0 : let target_file_size = self.get_checkpoint_distance();
2411 :
2412 : // Find the top of the historical layers
2413 0 : let end_lsn = {
2414 0 : let guard = self.layers.read().await;
2415 0 : let layers = guard.layer_map()?;
2416 :
2417 0 : let l0_deltas = layers.level0_deltas();
2418 0 :
2419 0 : // As an optimization, if we find that there are too few L0 layers,
2420 0 : // bail out early. We know that the compaction algorithm would do
2421 0 : // nothing in that case.
2422 0 : if l0_deltas.len() < fanout as usize {
2423 : // doesn't need compacting
2424 0 : return Ok(());
2425 0 : }
2426 0 : l0_deltas.iter().map(|l| l.lsn_range.end).max().unwrap()
2427 0 : };
2428 0 :
2429 0 : // Is the timeline being deleted?
2430 0 : if self.is_stopping() {
2431 0 : trace!("Dropping out of compaction on timeline shutdown");
2432 0 : return Err(CompactionError::ShuttingDown);
2433 0 : }
2434 :
2435 0 : let (dense_ks, _sparse_ks) = self.collect_keyspace(end_lsn, ctx).await?;
2436 : // TODO(chi): ignore sparse_keyspace for now, compact it in the future.
2437 0 : let mut adaptor = TimelineAdaptor::new(self, (end_lsn, dense_ks));
2438 0 :
2439 0 : pageserver_compaction::compact_tiered::compact_tiered(
2440 0 : &mut adaptor,
2441 0 : end_lsn,
2442 0 : target_file_size,
2443 0 : fanout,
2444 0 : ctx,
2445 0 : )
2446 0 : .await
2447 : // TODO: compact_tiered needs to return CompactionError
2448 0 : .map_err(CompactionError::Other)?;
2449 :
2450 0 : adaptor.flush_updates().await?;
2451 0 : Ok(())
2452 0 : }
2453 :
2454 : /// Take a list of images and deltas, produce images and deltas according to GC horizon and retain_lsns.
2455 : ///
2456 : /// It takes a key, the values of the key within the compaction process, a GC horizon, and all retain_lsns below the horizon.
2457 : /// For now, it requires the `accumulated_values` contains the full history of the key (i.e., the key with the lowest LSN is
2458 : /// an image or a WAL not requiring a base image). This restriction will be removed once we implement gc-compaction on branch.
2459 : ///
2460 : /// The function returns the deltas and the base image that need to be placed at each of the retain LSN. For example, we have:
2461 : ///
2462 : /// A@0x10, +B@0x20, +C@0x30, +D@0x40, +E@0x50, +F@0x60
2463 : /// horizon = 0x50, retain_lsn = 0x20, 0x40, delta_threshold=3
2464 : ///
2465 : /// The function will produce:
2466 : ///
2467 : /// ```plain
2468 : /// 0x20(retain_lsn) -> img=AB@0x20 always produce a single image below the lowest retain LSN
2469 : /// 0x40(retain_lsn) -> deltas=[+C@0x30, +D@0x40] two deltas since the last base image, keeping the deltas
2470 : /// 0x50(horizon) -> deltas=[ABCDE@0x50] three deltas since the last base image, generate an image but put it in the delta
2471 : /// above_horizon -> deltas=[+F@0x60] full history above the horizon
2472 : /// ```
2473 : ///
2474 : /// Note that `accumulated_values` must be sorted by LSN and should belong to a single key.
2475 : #[allow(clippy::too_many_arguments)]
2476 3888 : pub(crate) async fn generate_key_retention(
2477 3888 : self: &Arc<Timeline>,
2478 3888 : key: Key,
2479 3888 : full_history: &[(Key, Lsn, Value)],
2480 3888 : horizon: Lsn,
2481 3888 : retain_lsn_below_horizon: &[Lsn],
2482 3888 : delta_threshold_cnt: usize,
2483 3888 : base_img_from_ancestor: Option<(Key, Lsn, Bytes)>,
2484 3888 : verification: bool,
2485 3888 : ) -> anyhow::Result<KeyHistoryRetention> {
2486 : // Pre-checks for the invariants
2487 :
2488 3888 : let debug_mode = cfg!(debug_assertions) || cfg!(feature = "testing");
2489 :
2490 3888 : if debug_mode {
2491 9432 : for (log_key, _, _) in full_history {
2492 5544 : assert_eq!(log_key, &key, "mismatched key");
2493 : }
2494 3888 : for i in 1..full_history.len() {
2495 1656 : assert!(full_history[i - 1].1 <= full_history[i].1, "unordered LSN");
2496 1656 : if full_history[i - 1].1 == full_history[i].1 {
2497 0 : assert!(
2498 0 : matches!(full_history[i - 1].2, Value::Image(_)),
2499 0 : "unordered delta/image, or duplicated delta"
2500 : );
2501 1656 : }
2502 : }
2503 : // There was an assertion for no base image that checks if the first
2504 : // record in the history is `will_init` before, but it was removed.
2505 : // This is explained in the test cases for generate_key_retention.
2506 : // Search "incomplete history" for more information.
2507 8568 : for lsn in retain_lsn_below_horizon {
2508 4680 : assert!(lsn < &horizon, "retain lsn must be below horizon")
2509 : }
2510 3888 : for i in 1..retain_lsn_below_horizon.len() {
2511 2136 : assert!(
2512 2136 : retain_lsn_below_horizon[i - 1] <= retain_lsn_below_horizon[i],
2513 0 : "unordered LSN"
2514 : );
2515 : }
2516 0 : }
2517 3888 : let has_ancestor = base_img_from_ancestor.is_some();
2518 : // Step 1: split history into len(retain_lsn_below_horizon) + 2 buckets, where the last bucket is for all deltas above the horizon,
2519 : // and the second-to-last bucket is for the horizon. Each bucket contains lsn_last_bucket < deltas <= lsn_this_bucket.
2520 3888 : let (mut split_history, lsn_split_points) = {
2521 3888 : let mut split_history = Vec::new();
2522 3888 : split_history.resize_with(retain_lsn_below_horizon.len() + 2, Vec::new);
2523 3888 : let mut lsn_split_points = Vec::with_capacity(retain_lsn_below_horizon.len() + 1);
2524 8568 : for lsn in retain_lsn_below_horizon {
2525 4680 : lsn_split_points.push(*lsn);
2526 4680 : }
2527 3888 : lsn_split_points.push(horizon);
2528 3888 : let mut current_idx = 0;
2529 9432 : for item @ (_, lsn, _) in full_history {
2530 7008 : while current_idx < lsn_split_points.len() && *lsn > lsn_split_points[current_idx] {
2531 1464 : current_idx += 1;
2532 1464 : }
2533 5544 : split_history[current_idx].push(item);
2534 : }
2535 3888 : (split_history, lsn_split_points)
2536 : };
2537 : // Step 2: filter out duplicated records due to the k-merge of image/delta layers
2538 16344 : for split_for_lsn in &mut split_history {
2539 12456 : let mut prev_lsn = None;
2540 12456 : let mut new_split_for_lsn = Vec::with_capacity(split_for_lsn.len());
2541 12456 : for record @ (_, lsn, _) in std::mem::take(split_for_lsn) {
2542 5544 : if let Some(prev_lsn) = &prev_lsn {
2543 744 : if *prev_lsn == lsn {
2544 : // The case that we have an LSN with both data from the delta layer and the image layer. As
2545 : // `ValueWrapper` ensures that an image is ordered before a delta at the same LSN, we simply
2546 : // drop this delta and keep the image.
2547 : //
2548 : // For example, we have delta layer key1@0x10, key1@0x20, and image layer key1@0x10, we will
2549 : // keep the image for key1@0x10 and the delta for key1@0x20. key1@0x10 delta will be simply
2550 : // dropped.
2551 : //
2552 : // TODO: in case we have both delta + images for a given LSN and it does not exceed the delta
2553 : // threshold, we could have kept delta instead to save space. This is an optimization for the future.
2554 0 : continue;
2555 744 : }
2556 4800 : }
2557 5544 : prev_lsn = Some(lsn);
2558 5544 : new_split_for_lsn.push(record);
2559 : }
2560 12456 : *split_for_lsn = new_split_for_lsn;
2561 : }
2562 : // Step 3: generate images when necessary
2563 3888 : let mut retention = Vec::with_capacity(split_history.len());
2564 3888 : let mut records_since_last_image = 0;
2565 3888 : let batch_cnt = split_history.len();
2566 3888 : assert!(
2567 3888 : batch_cnt >= 2,
2568 0 : "should have at least below + above horizon batches"
2569 : );
2570 3888 : let mut replay_history: Vec<(Key, Lsn, Value)> = Vec::new();
2571 3888 : if let Some((key, lsn, ref img)) = base_img_from_ancestor {
2572 252 : replay_history.push((key, lsn, Value::Image(img.clone())));
2573 3636 : }
2574 :
2575 : /// Generate debug information for the replay history
2576 0 : fn generate_history_trace(replay_history: &[(Key, Lsn, Value)]) -> String {
2577 : use std::fmt::Write;
2578 0 : let mut output = String::new();
2579 0 : if let Some((key, _, _)) = replay_history.first() {
2580 0 : write!(output, "key={} ", key).unwrap();
2581 0 : let mut cnt = 0;
2582 0 : for (_, lsn, val) in replay_history {
2583 0 : if val.is_image() {
2584 0 : write!(output, "i@{} ", lsn).unwrap();
2585 0 : } else if val.will_init() {
2586 0 : write!(output, "di@{} ", lsn).unwrap();
2587 0 : } else {
2588 0 : write!(output, "d@{} ", lsn).unwrap();
2589 0 : }
2590 0 : cnt += 1;
2591 0 : if cnt >= 128 {
2592 0 : write!(output, "... and more").unwrap();
2593 0 : break;
2594 0 : }
2595 : }
2596 0 : } else {
2597 0 : write!(output, "<no history>").unwrap();
2598 0 : }
2599 0 : output
2600 0 : }
2601 :
2602 0 : fn generate_debug_trace(
2603 0 : replay_history: Option<&[(Key, Lsn, Value)]>,
2604 0 : full_history: &[(Key, Lsn, Value)],
2605 0 : lsns: &[Lsn],
2606 0 : horizon: Lsn,
2607 0 : ) -> String {
2608 : use std::fmt::Write;
2609 0 : let mut output = String::new();
2610 0 : if let Some(replay_history) = replay_history {
2611 0 : writeln!(
2612 0 : output,
2613 0 : "replay_history: {}",
2614 0 : generate_history_trace(replay_history)
2615 0 : )
2616 0 : .unwrap();
2617 0 : } else {
2618 0 : writeln!(output, "replay_history: <disabled>",).unwrap();
2619 0 : }
2620 0 : writeln!(
2621 0 : output,
2622 0 : "full_history: {}",
2623 0 : generate_history_trace(full_history)
2624 0 : )
2625 0 : .unwrap();
2626 0 : writeln!(
2627 0 : output,
2628 0 : "when processing: [{}] horizon={}",
2629 0 : lsns.iter().map(|l| format!("{l}")).join(","),
2630 0 : horizon
2631 0 : )
2632 0 : .unwrap();
2633 0 : output
2634 0 : }
2635 :
2636 3888 : let mut key_exists = false;
2637 12444 : for (i, split_for_lsn) in split_history.into_iter().enumerate() {
2638 : // TODO: there could be image keys inside the splits, and we can compute records_since_last_image accordingly.
2639 12444 : records_since_last_image += split_for_lsn.len();
2640 : // Whether to produce an image into the final layer files
2641 12444 : let produce_image = if i == 0 && !has_ancestor {
2642 : // We always generate images for the first batch (below horizon / lowest retain_lsn)
2643 3636 : true
2644 8808 : } else if i == batch_cnt - 1 {
2645 : // Do not generate images for the last batch (above horizon)
2646 3876 : false
2647 4932 : } else if records_since_last_image == 0 {
2648 3864 : false
2649 1068 : } else if records_since_last_image >= delta_threshold_cnt {
2650 : // Generate images when there are too many records
2651 36 : true
2652 : } else {
2653 1032 : false
2654 : };
2655 12444 : replay_history.extend(split_for_lsn.iter().map(|x| (*x).clone()));
2656 : // Only retain the items after the last image record
2657 15324 : for idx in (0..replay_history.len()).rev() {
2658 15324 : if replay_history[idx].2.will_init() {
2659 12444 : replay_history = replay_history[idx..].to_vec();
2660 12444 : break;
2661 2880 : }
2662 : }
2663 12444 : if replay_history.is_empty() && !key_exists {
2664 : // The key does not exist at earlier LSN, we can skip this iteration.
2665 0 : retention.push(Vec::new());
2666 0 : continue;
2667 12444 : } else {
2668 12444 : key_exists = true;
2669 12444 : }
2670 12444 : let Some((_, _, val)) = replay_history.first() else {
2671 0 : unreachable!("replay history should not be empty once it exists")
2672 : };
2673 12444 : if !val.will_init() {
2674 0 : return Err(anyhow::anyhow!("invalid history, no base image")).with_context(|| {
2675 0 : generate_debug_trace(
2676 0 : Some(&replay_history),
2677 0 : full_history,
2678 0 : retain_lsn_below_horizon,
2679 0 : horizon,
2680 0 : )
2681 0 : });
2682 12444 : }
2683 : // Whether to reconstruct the image. In debug mode, we will generate an image
2684 : // at every retain_lsn to ensure data is not corrupted, but we won't put the
2685 : // image into the final layer.
2686 12444 : let img_and_lsn = if produce_image {
2687 3672 : records_since_last_image = 0;
2688 3672 : let replay_history_for_debug = if debug_mode {
2689 3672 : Some(replay_history.clone())
2690 : } else {
2691 0 : None
2692 : };
2693 3672 : let replay_history_for_debug_ref = replay_history_for_debug.as_deref();
2694 3672 : let history = std::mem::take(&mut replay_history);
2695 3672 : let mut img = None;
2696 3672 : let mut records = Vec::with_capacity(history.len());
2697 3672 : if let (_, lsn, Value::Image(val)) = history.first().as_ref().unwrap() {
2698 3540 : img = Some((*lsn, val.clone()));
2699 3540 : for (_, lsn, val) in history.into_iter().skip(1) {
2700 240 : let Value::WalRecord(rec) = val else {
2701 0 : return Err(anyhow::anyhow!(
2702 0 : "invalid record, first record is image, expect walrecords"
2703 0 : ))
2704 0 : .with_context(|| {
2705 0 : generate_debug_trace(
2706 0 : replay_history_for_debug_ref,
2707 0 : full_history,
2708 0 : retain_lsn_below_horizon,
2709 0 : horizon,
2710 0 : )
2711 0 : });
2712 : };
2713 240 : records.push((lsn, rec));
2714 : }
2715 : } else {
2716 216 : for (_, lsn, val) in history.into_iter() {
2717 216 : let Value::WalRecord(rec) = val else {
2718 0 : return Err(anyhow::anyhow!("invalid record, first record is walrecord, expect rest are walrecord"))
2719 0 : .with_context(|| generate_debug_trace(
2720 0 : replay_history_for_debug_ref,
2721 0 : full_history,
2722 0 : retain_lsn_below_horizon,
2723 0 : horizon,
2724 0 : ));
2725 : };
2726 216 : records.push((lsn, rec));
2727 : }
2728 : }
2729 : // WAL redo requires records in the reverse LSN order
2730 3672 : records.reverse();
2731 3672 : let state = ValueReconstructState { img, records };
2732 : // last batch does not generate image so i is always in range, unless we force generate
2733 : // an image during testing
2734 3672 : let request_lsn = if i >= lsn_split_points.len() {
2735 0 : Lsn::MAX
2736 : } else {
2737 3672 : lsn_split_points[i]
2738 : };
2739 3672 : let img = self
2740 3672 : .reconstruct_value(key, request_lsn, state, RedoAttemptType::GcCompaction)
2741 3672 : .await?;
2742 3660 : Some((request_lsn, img))
2743 : } else {
2744 8772 : None
2745 : };
2746 12432 : if produce_image {
2747 3660 : let (request_lsn, img) = img_and_lsn.unwrap();
2748 3660 : replay_history.push((key, request_lsn, Value::Image(img.clone())));
2749 3660 : retention.push(vec![(request_lsn, Value::Image(img))]);
2750 8772 : } else {
2751 8772 : let deltas = split_for_lsn
2752 8772 : .iter()
2753 8772 : .map(|(_, lsn, value)| (*lsn, value.clone()))
2754 8772 : .collect_vec();
2755 8772 : retention.push(deltas);
2756 8772 : }
2757 : }
2758 3876 : let mut result = Vec::with_capacity(retention.len());
2759 3876 : assert_eq!(retention.len(), lsn_split_points.len() + 1);
2760 12432 : for (idx, logs) in retention.into_iter().enumerate() {
2761 12432 : if idx == lsn_split_points.len() {
2762 3876 : let retention = KeyHistoryRetention {
2763 3876 : below_horizon: result,
2764 3876 : above_horizon: KeyLogAtLsn(logs),
2765 3876 : };
2766 3876 : if verification {
2767 3876 : retention
2768 3876 : .verify(key, &base_img_from_ancestor, full_history, self)
2769 3876 : .await?;
2770 0 : }
2771 3876 : return Ok(retention);
2772 8556 : } else {
2773 8556 : result.push((lsn_split_points[idx], KeyLogAtLsn(logs)));
2774 8556 : }
2775 : }
2776 0 : unreachable!("key retention is empty")
2777 3888 : }
2778 :
2779 : /// Check how much space is left on the disk
2780 324 : async fn check_available_space(self: &Arc<Self>) -> anyhow::Result<u64> {
2781 324 : let tenants_dir = self.conf.tenants_path();
2782 :
2783 324 : let stat = Statvfs::get(&tenants_dir, None)
2784 324 : .context("statvfs failed, presumably directory got unlinked")?;
2785 :
2786 324 : let (avail_bytes, _) = stat.get_avail_total_bytes();
2787 324 :
2788 324 : Ok(avail_bytes)
2789 324 : }
2790 :
2791 : /// Check if the compaction can proceed safely without running out of space. We assume the size
2792 : /// upper bound of the produced files of a compaction job is the same as all layers involved in
2793 : /// the compaction. Therefore, we need `2 * layers_to_be_compacted_size` at least to do a
2794 : /// compaction.
2795 324 : async fn check_compaction_space(
2796 324 : self: &Arc<Self>,
2797 324 : layer_selection: &[Layer],
2798 324 : ) -> Result<(), CompactionError> {
2799 324 : let available_space = self
2800 324 : .check_available_space()
2801 324 : .await
2802 324 : .map_err(CompactionError::Other)?;
2803 324 : let mut remote_layer_size = 0;
2804 324 : let mut all_layer_size = 0;
2805 1272 : for layer in layer_selection {
2806 948 : let needs_download = layer
2807 948 : .needs_download()
2808 948 : .await
2809 948 : .context("failed to check if layer needs download")
2810 948 : .map_err(CompactionError::Other)?;
2811 948 : if needs_download.is_some() {
2812 0 : remote_layer_size += layer.layer_desc().file_size;
2813 948 : }
2814 948 : all_layer_size += layer.layer_desc().file_size;
2815 : }
2816 324 : let allocated_space = (available_space as f64 * 0.8) as u64; /* reserve 20% space for other tasks */
2817 324 : if all_layer_size /* space needed for newly-generated file */ + remote_layer_size /* space for downloading layers */ > allocated_space
2818 : {
2819 0 : return Err(CompactionError::Other(anyhow!(
2820 0 : "not enough space for compaction: available_space={}, allocated_space={}, all_layer_size={}, remote_layer_size={}, required_space={}",
2821 0 : available_space,
2822 0 : allocated_space,
2823 0 : all_layer_size,
2824 0 : remote_layer_size,
2825 0 : all_layer_size + remote_layer_size
2826 0 : )));
2827 324 : }
2828 324 : Ok(())
2829 324 : }
2830 :
2831 : /// Check if the memory usage is within the limit.
2832 324 : async fn check_memory_usage(
2833 324 : self: &Arc<Self>,
2834 324 : layer_selection: &[Layer],
2835 324 : ) -> Result<(), CompactionError> {
2836 324 : let mut estimated_memory_usage_mb = 0.0;
2837 324 : let mut num_image_layers = 0;
2838 324 : let mut num_delta_layers = 0;
2839 324 : let target_layer_size_bytes = 256 * 1024 * 1024;
2840 1272 : for layer in layer_selection {
2841 948 : let layer_desc = layer.layer_desc();
2842 948 : if layer_desc.is_delta() {
2843 528 : // Delta layers at most have 1MB buffer; 3x to make it safe (there're deltas as large as 16KB).
2844 528 : // Multiply the layer size so that tests can pass.
2845 528 : estimated_memory_usage_mb +=
2846 528 : 3.0 * (layer_desc.file_size / target_layer_size_bytes) as f64;
2847 528 : num_delta_layers += 1;
2848 528 : } else {
2849 420 : // Image layers at most have 1MB buffer but it might be compressed; assume 5x compression ratio.
2850 420 : estimated_memory_usage_mb +=
2851 420 : 5.0 * (layer_desc.file_size / target_layer_size_bytes) as f64;
2852 420 : num_image_layers += 1;
2853 420 : }
2854 : }
2855 324 : if estimated_memory_usage_mb > 1024.0 {
2856 0 : return Err(CompactionError::Other(anyhow!(
2857 0 : "estimated memory usage is too high: {}MB, giving up compaction; num_image_layers={}, num_delta_layers={}",
2858 0 : estimated_memory_usage_mb,
2859 0 : num_image_layers,
2860 0 : num_delta_layers
2861 0 : )));
2862 324 : }
2863 324 : Ok(())
2864 324 : }
2865 :
2866 : /// Get a watermark for gc-compaction, that is the lowest LSN that we can use as the `gc_horizon` for
2867 : /// the compaction algorithm. It is min(space_cutoff, time_cutoff, latest_gc_cutoff, standby_horizon).
2868 : /// Leases and retain_lsns are considered in the gc-compaction job itself so we don't need to account for them
2869 : /// here.
2870 336 : pub(crate) fn get_gc_compaction_watermark(self: &Arc<Self>) -> Lsn {
2871 336 : let gc_cutoff_lsn = {
2872 336 : let gc_info = self.gc_info.read().unwrap();
2873 336 : gc_info.min_cutoff()
2874 336 : };
2875 336 :
2876 336 : // TODO: standby horizon should use leases so we don't really need to consider it here.
2877 336 : // let watermark = watermark.min(self.standby_horizon.load());
2878 336 :
2879 336 : // TODO: ensure the child branches will not use anything below the watermark, or consider
2880 336 : // them when computing the watermark.
2881 336 : gc_cutoff_lsn.min(*self.get_applied_gc_cutoff_lsn())
2882 336 : }
2883 :
2884 : /// Split a gc-compaction job into multiple compaction jobs. The split is based on the key range and the estimated size of the compaction job.
2885 : /// The function returns a list of compaction jobs that can be executed separately. If the upper bound of the compact LSN
2886 : /// range is not specified, we will use the latest gc_cutoff as the upper bound, so that all jobs in the jobset acts
2887 : /// like a full compaction of the specified keyspace.
2888 0 : pub(crate) async fn gc_compaction_split_jobs(
2889 0 : self: &Arc<Self>,
2890 0 : job: GcCompactJob,
2891 0 : sub_compaction_max_job_size_mb: Option<u64>,
2892 0 : ) -> Result<Vec<GcCompactJob>, CompactionError> {
2893 0 : let compact_below_lsn = if job.compact_lsn_range.end != Lsn::MAX {
2894 0 : job.compact_lsn_range.end
2895 : } else {
2896 0 : self.get_gc_compaction_watermark()
2897 : };
2898 :
2899 0 : if compact_below_lsn == Lsn::INVALID {
2900 0 : tracing::warn!(
2901 0 : "no layers to compact with gc: gc_cutoff not generated yet, skipping gc bottom-most compaction"
2902 : );
2903 0 : return Ok(vec![]);
2904 0 : }
2905 :
2906 : // Split compaction job to about 4GB each
2907 : const GC_COMPACT_MAX_SIZE_MB: u64 = 4 * 1024;
2908 0 : let sub_compaction_max_job_size_mb =
2909 0 : sub_compaction_max_job_size_mb.unwrap_or(GC_COMPACT_MAX_SIZE_MB);
2910 0 :
2911 0 : let mut compact_jobs = Vec::<GcCompactJob>::new();
2912 0 : // For now, we simply use the key partitioning information; we should do a more fine-grained partitioning
2913 0 : // by estimating the amount of files read for a compaction job. We should also partition on LSN.
2914 0 : let ((dense_ks, sparse_ks), _) = self.partitioning.read().as_ref().clone();
2915 : // Truncate the key range to be within user specified compaction range.
2916 0 : fn truncate_to(
2917 0 : source_start: &Key,
2918 0 : source_end: &Key,
2919 0 : target_start: &Key,
2920 0 : target_end: &Key,
2921 0 : ) -> Option<(Key, Key)> {
2922 0 : let start = source_start.max(target_start);
2923 0 : let end = source_end.min(target_end);
2924 0 : if start < end {
2925 0 : Some((*start, *end))
2926 : } else {
2927 0 : None
2928 : }
2929 0 : }
2930 0 : let mut split_key_ranges = Vec::new();
2931 0 : let ranges = dense_ks
2932 0 : .parts
2933 0 : .iter()
2934 0 : .map(|partition| partition.ranges.iter())
2935 0 : .chain(sparse_ks.parts.iter().map(|x| x.0.ranges.iter()))
2936 0 : .flatten()
2937 0 : .cloned()
2938 0 : .collect_vec();
2939 0 : for range in ranges.iter() {
2940 0 : let Some((start, end)) = truncate_to(
2941 0 : &range.start,
2942 0 : &range.end,
2943 0 : &job.compact_key_range.start,
2944 0 : &job.compact_key_range.end,
2945 0 : ) else {
2946 0 : continue;
2947 : };
2948 0 : split_key_ranges.push((start, end));
2949 : }
2950 0 : split_key_ranges.sort();
2951 0 : let all_layers = {
2952 0 : let guard = self.layers.read().await;
2953 0 : let layer_map = guard.layer_map()?;
2954 0 : layer_map.iter_historic_layers().collect_vec()
2955 0 : };
2956 0 : let mut current_start = None;
2957 0 : let ranges_num = split_key_ranges.len();
2958 0 : for (idx, (start, end)) in split_key_ranges.into_iter().enumerate() {
2959 0 : if current_start.is_none() {
2960 0 : current_start = Some(start);
2961 0 : }
2962 0 : let start = current_start.unwrap();
2963 0 : if start >= end {
2964 : // We have already processed this partition.
2965 0 : continue;
2966 0 : }
2967 0 : let overlapping_layers = {
2968 0 : let mut desc = Vec::new();
2969 0 : for layer in all_layers.iter() {
2970 0 : if overlaps_with(&layer.get_key_range(), &(start..end))
2971 0 : && layer.get_lsn_range().start <= compact_below_lsn
2972 0 : {
2973 0 : desc.push(layer.clone());
2974 0 : }
2975 : }
2976 0 : desc
2977 0 : };
2978 0 : let total_size = overlapping_layers.iter().map(|x| x.file_size).sum::<u64>();
2979 0 : if total_size > sub_compaction_max_job_size_mb * 1024 * 1024 || ranges_num == idx + 1 {
2980 : // Try to extend the compaction range so that we include at least one full layer file.
2981 0 : let extended_end = overlapping_layers
2982 0 : .iter()
2983 0 : .map(|layer| layer.key_range.end)
2984 0 : .min();
2985 : // It is possible that the search range does not contain any layer files when we reach the end of the loop.
2986 : // In this case, we simply use the specified key range end.
2987 0 : let end = if let Some(extended_end) = extended_end {
2988 0 : extended_end.max(end)
2989 : } else {
2990 0 : end
2991 : };
2992 0 : let end = if ranges_num == idx + 1 {
2993 : // extend the compaction range to the end of the key range if it's the last partition
2994 0 : end.max(job.compact_key_range.end)
2995 : } else {
2996 0 : end
2997 : };
2998 0 : if total_size == 0 && !compact_jobs.is_empty() {
2999 0 : info!(
3000 0 : "splitting compaction job: {}..{}, estimated_size={}, extending the previous job",
3001 : start, end, total_size
3002 : );
3003 0 : compact_jobs.last_mut().unwrap().compact_key_range.end = end;
3004 0 : current_start = Some(end);
3005 : } else {
3006 0 : info!(
3007 0 : "splitting compaction job: {}..{}, estimated_size={}",
3008 : start, end, total_size
3009 : );
3010 0 : compact_jobs.push(GcCompactJob {
3011 0 : dry_run: job.dry_run,
3012 0 : compact_key_range: start..end,
3013 0 : compact_lsn_range: job.compact_lsn_range.start..compact_below_lsn,
3014 0 : });
3015 0 : current_start = Some(end);
3016 : }
3017 0 : }
3018 : }
3019 0 : Ok(compact_jobs)
3020 0 : }
3021 :
3022 : /// An experimental compaction building block that combines compaction with garbage collection.
3023 : ///
3024 : /// The current implementation picks all delta + image layers that are below or intersecting with
3025 : /// the GC horizon without considering retain_lsns. Then, it does a full compaction over all these delta
3026 : /// layers and image layers, which generates image layers on the gc horizon, drop deltas below gc horizon,
3027 : /// and create delta layers with all deltas >= gc horizon.
3028 : ///
3029 : /// If `options.compact_range` is provided, it will only compact the keys within the range, aka partial compaction.
3030 : /// Partial compaction will read and process all layers overlapping with the key range, even if it might
3031 : /// contain extra keys. After the gc-compaction phase completes, delta layers that are not fully contained
3032 : /// within the key range will be rewritten to ensure they do not overlap with the delta layers. Providing
3033 : /// Key::MIN..Key..MAX to the function indicates a full compaction, though technically, `Key::MAX` is not
3034 : /// part of the range.
3035 : ///
3036 : /// If `options.compact_lsn_range.end` is provided, the compaction will only compact layers below or intersect with
3037 : /// the LSN. Otherwise, it will use the gc cutoff by default.
3038 336 : pub(crate) async fn compact_with_gc(
3039 336 : self: &Arc<Self>,
3040 336 : cancel: &CancellationToken,
3041 336 : options: CompactOptions,
3042 336 : ctx: &RequestContext,
3043 336 : ) -> Result<CompactionOutcome, CompactionError> {
3044 336 : let sub_compaction = options.sub_compaction;
3045 336 : let job = GcCompactJob::from_compact_options(options.clone());
3046 336 : let yield_for_l0 = options.flags.contains(CompactFlags::YieldForL0);
3047 336 : if sub_compaction {
3048 0 : info!(
3049 0 : "running enhanced gc bottom-most compaction with sub-compaction, splitting compaction jobs"
3050 : );
3051 0 : let jobs = self
3052 0 : .gc_compaction_split_jobs(job, options.sub_compaction_max_job_size_mb)
3053 0 : .await?;
3054 0 : let jobs_len = jobs.len();
3055 0 : for (idx, job) in jobs.into_iter().enumerate() {
3056 0 : info!(
3057 0 : "running enhanced gc bottom-most compaction, sub-compaction {}/{}",
3058 0 : idx + 1,
3059 : jobs_len
3060 : );
3061 0 : self.compact_with_gc_inner(cancel, job, ctx, yield_for_l0)
3062 0 : .await?;
3063 : }
3064 0 : if jobs_len == 0 {
3065 0 : info!("no jobs to run, skipping gc bottom-most compaction");
3066 0 : }
3067 0 : return Ok(CompactionOutcome::Done);
3068 336 : }
3069 336 : self.compact_with_gc_inner(cancel, job, ctx, yield_for_l0)
3070 336 : .await
3071 336 : }
3072 :
3073 336 : async fn compact_with_gc_inner(
3074 336 : self: &Arc<Self>,
3075 336 : cancel: &CancellationToken,
3076 336 : job: GcCompactJob,
3077 336 : ctx: &RequestContext,
3078 336 : yield_for_l0: bool,
3079 336 : ) -> Result<CompactionOutcome, CompactionError> {
3080 336 : // Block other compaction/GC tasks from running for now. GC-compaction could run along
3081 336 : // with legacy compaction tasks in the future. Always ensure the lock order is compaction -> gc.
3082 336 : // Note that we already acquired the compaction lock when the outer `compact` function gets called.
3083 336 :
3084 336 : let timer = Instant::now();
3085 336 : let begin_timer = timer;
3086 336 :
3087 336 : let gc_lock = async {
3088 336 : tokio::select! {
3089 336 : guard = self.gc_lock.lock() => Ok(guard),
3090 336 : _ = cancel.cancelled() => Err(CompactionError::ShuttingDown),
3091 : }
3092 336 : };
3093 :
3094 336 : let time_acquire_lock = timer.elapsed();
3095 336 : let timer = Instant::now();
3096 :
3097 336 : let gc_lock = crate::timed(
3098 336 : gc_lock,
3099 336 : "acquires gc lock",
3100 336 : std::time::Duration::from_secs(5),
3101 336 : )
3102 336 : .await?;
3103 :
3104 336 : let dry_run = job.dry_run;
3105 336 : let compact_key_range = job.compact_key_range;
3106 336 : let compact_lsn_range = job.compact_lsn_range;
3107 :
3108 336 : let debug_mode = cfg!(debug_assertions) || cfg!(feature = "testing");
3109 :
3110 336 : info!(
3111 0 : "running enhanced gc bottom-most compaction, dry_run={dry_run}, compact_key_range={}..{}, compact_lsn_range={}..{}",
3112 : compact_key_range.start,
3113 : compact_key_range.end,
3114 : compact_lsn_range.start,
3115 : compact_lsn_range.end
3116 : );
3117 :
3118 336 : scopeguard::defer! {
3119 336 : info!("done enhanced gc bottom-most compaction");
3120 336 : };
3121 336 :
3122 336 : let mut stat = CompactionStatistics::default();
3123 :
3124 : // Step 0: pick all delta layers + image layers below/intersect with the GC horizon.
3125 : // The layer selection has the following properties:
3126 : // 1. If a layer is in the selection, all layers below it are in the selection.
3127 : // 2. Inferred from (1), for each key in the layer selection, the value can be reconstructed only with the layers in the layer selection.
3128 324 : let job_desc = {
3129 336 : let guard = self.layers.read().await;
3130 336 : let layers = guard.layer_map()?;
3131 336 : let gc_info = self.gc_info.read().unwrap();
3132 336 : let mut retain_lsns_below_horizon = Vec::new();
3133 336 : let gc_cutoff = {
3134 : // Currently, gc-compaction only kicks in after the legacy gc has updated the gc_cutoff.
3135 : // Therefore, it can only clean up data that cannot be cleaned up with legacy gc, instead of
3136 : // cleaning everything that theoritically it could. In the future, it should use `self.gc_info`
3137 : // to get the truth data.
3138 336 : let real_gc_cutoff = self.get_gc_compaction_watermark();
3139 : // The compaction algorithm will keep all keys above the gc_cutoff while keeping only necessary keys below the gc_cutoff for
3140 : // each of the retain_lsn. Therefore, if the user-provided `compact_lsn_range.end` is larger than the real gc cutoff, we will use
3141 : // the real cutoff.
3142 336 : let mut gc_cutoff = if compact_lsn_range.end == Lsn::MAX {
3143 300 : if real_gc_cutoff == Lsn::INVALID {
3144 : // If the gc_cutoff is not generated yet, we should not compact anything.
3145 0 : tracing::warn!(
3146 0 : "no layers to compact with gc: gc_cutoff not generated yet, skipping gc bottom-most compaction"
3147 : );
3148 0 : return Ok(CompactionOutcome::Skipped);
3149 300 : }
3150 300 : real_gc_cutoff
3151 : } else {
3152 36 : compact_lsn_range.end
3153 : };
3154 336 : if gc_cutoff > real_gc_cutoff {
3155 24 : warn!(
3156 0 : "provided compact_lsn_range.end={} is larger than the real_gc_cutoff={}, using the real gc cutoff",
3157 : gc_cutoff, real_gc_cutoff
3158 : );
3159 24 : gc_cutoff = real_gc_cutoff;
3160 312 : }
3161 336 : gc_cutoff
3162 : };
3163 420 : for (lsn, _timeline_id, _is_offloaded) in &gc_info.retain_lsns {
3164 420 : if lsn < &gc_cutoff {
3165 420 : retain_lsns_below_horizon.push(*lsn);
3166 420 : }
3167 : }
3168 336 : for lsn in gc_info.leases.keys() {
3169 0 : if lsn < &gc_cutoff {
3170 0 : retain_lsns_below_horizon.push(*lsn);
3171 0 : }
3172 : }
3173 336 : let mut selected_layers: Vec<Layer> = Vec::new();
3174 336 : drop(gc_info);
3175 : // Firstly, pick all the layers intersect or below the gc_cutoff, get the largest LSN in the selected layers.
3176 336 : let Some(max_layer_lsn) = layers
3177 336 : .iter_historic_layers()
3178 1500 : .filter(|desc| desc.get_lsn_range().start <= gc_cutoff)
3179 1284 : .map(|desc| desc.get_lsn_range().end)
3180 336 : .max()
3181 : else {
3182 0 : info!(
3183 0 : "no layers to compact with gc: no historic layers below gc_cutoff, gc_cutoff={}",
3184 : gc_cutoff
3185 : );
3186 0 : return Ok(CompactionOutcome::Done);
3187 : };
3188 : // Next, if the user specifies compact_lsn_range.start, we need to filter some layers out. All the layers (strictly) below
3189 : // the min_layer_lsn computed as below will be filtered out and the data will be accessed using the normal read path, as if
3190 : // it is a branch.
3191 336 : let Some(min_layer_lsn) = layers
3192 336 : .iter_historic_layers()
3193 1500 : .filter(|desc| {
3194 1500 : if compact_lsn_range.start == Lsn::INVALID {
3195 1224 : true // select all layers below if start == Lsn(0)
3196 : } else {
3197 276 : desc.get_lsn_range().end > compact_lsn_range.start // strictly larger than compact_above_lsn
3198 : }
3199 1500 : })
3200 1392 : .map(|desc| desc.get_lsn_range().start)
3201 336 : .min()
3202 : else {
3203 0 : info!(
3204 0 : "no layers to compact with gc: no historic layers above compact_above_lsn, compact_above_lsn={}",
3205 : compact_lsn_range.end
3206 : );
3207 0 : return Ok(CompactionOutcome::Done);
3208 : };
3209 : // Then, pick all the layers that are below the max_layer_lsn. This is to ensure we can pick all single-key
3210 : // layers to compact.
3211 336 : let mut rewrite_layers = Vec::new();
3212 1500 : for desc in layers.iter_historic_layers() {
3213 1500 : if desc.get_lsn_range().end <= max_layer_lsn
3214 1284 : && desc.get_lsn_range().start >= min_layer_lsn
3215 1176 : && overlaps_with(&desc.get_key_range(), &compact_key_range)
3216 : {
3217 : // If the layer overlaps with the compaction key range, we need to read it to obtain all keys within the range,
3218 : // even if it might contain extra keys
3219 948 : selected_layers.push(guard.get_from_desc(&desc));
3220 948 : // 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
3221 948 : // to overlap image layers)
3222 948 : if desc.is_delta() && !fully_contains(&compact_key_range, &desc.get_key_range())
3223 12 : {
3224 12 : rewrite_layers.push(desc);
3225 936 : }
3226 552 : }
3227 : }
3228 336 : if selected_layers.is_empty() {
3229 12 : info!(
3230 0 : "no layers to compact with gc: no layers within the key range, gc_cutoff={}, key_range={}..{}",
3231 : gc_cutoff, compact_key_range.start, compact_key_range.end
3232 : );
3233 12 : return Ok(CompactionOutcome::Done);
3234 324 : }
3235 324 : retain_lsns_below_horizon.sort();
3236 324 : GcCompactionJobDescription {
3237 324 : selected_layers,
3238 324 : gc_cutoff,
3239 324 : retain_lsns_below_horizon,
3240 324 : min_layer_lsn,
3241 324 : max_layer_lsn,
3242 324 : compaction_key_range: compact_key_range,
3243 324 : rewrite_layers,
3244 324 : }
3245 : };
3246 324 : let (has_data_below, lowest_retain_lsn) = if compact_lsn_range.start != Lsn::INVALID {
3247 : // If we only compact above some LSN, we should get the history from the current branch below the specified LSN.
3248 : // We use job_desc.min_layer_lsn as if it's the lowest branch point.
3249 48 : (true, job_desc.min_layer_lsn)
3250 276 : } else if self.ancestor_timeline.is_some() {
3251 : // In theory, we can also use min_layer_lsn here, but using ancestor LSN makes sure the delta layers cover the
3252 : // LSN ranges all the way to the ancestor timeline.
3253 12 : (true, self.ancestor_lsn)
3254 : } else {
3255 264 : let res = job_desc
3256 264 : .retain_lsns_below_horizon
3257 264 : .first()
3258 264 : .copied()
3259 264 : .unwrap_or(job_desc.gc_cutoff);
3260 264 : if debug_mode {
3261 264 : assert_eq!(
3262 264 : res,
3263 264 : job_desc
3264 264 : .retain_lsns_below_horizon
3265 264 : .iter()
3266 264 : .min()
3267 264 : .copied()
3268 264 : .unwrap_or(job_desc.gc_cutoff)
3269 264 : );
3270 0 : }
3271 264 : (false, res)
3272 : };
3273 :
3274 324 : let verification = self.get_gc_compaction_settings().gc_compaction_verification;
3275 324 :
3276 324 : info!(
3277 0 : "picked {} layers for compaction ({} layers need rewriting) with max_layer_lsn={} min_layer_lsn={} gc_cutoff={} lowest_retain_lsn={}, key_range={}..{}, has_data_below={}",
3278 0 : job_desc.selected_layers.len(),
3279 0 : job_desc.rewrite_layers.len(),
3280 : job_desc.max_layer_lsn,
3281 : job_desc.min_layer_lsn,
3282 : job_desc.gc_cutoff,
3283 : lowest_retain_lsn,
3284 : job_desc.compaction_key_range.start,
3285 : job_desc.compaction_key_range.end,
3286 : has_data_below,
3287 : );
3288 :
3289 324 : let time_analyze = timer.elapsed();
3290 324 : let timer = Instant::now();
3291 :
3292 1272 : for layer in &job_desc.selected_layers {
3293 948 : debug!("read layer: {}", layer.layer_desc().key());
3294 : }
3295 336 : for layer in &job_desc.rewrite_layers {
3296 12 : debug!("rewrite layer: {}", layer.key());
3297 : }
3298 :
3299 324 : self.check_compaction_space(&job_desc.selected_layers)
3300 324 : .await?;
3301 :
3302 324 : self.check_memory_usage(&job_desc.selected_layers).await?;
3303 324 : if job_desc.selected_layers.len() > 100
3304 0 : && job_desc.rewrite_layers.len() as f64 >= job_desc.selected_layers.len() as f64 * 0.7
3305 : {
3306 0 : return Err(CompactionError::Other(anyhow!(
3307 0 : "too many layers to rewrite: {} / {}, giving up compaction",
3308 0 : job_desc.rewrite_layers.len(),
3309 0 : job_desc.selected_layers.len()
3310 0 : )));
3311 324 : }
3312 :
3313 : // Generate statistics for the compaction
3314 1272 : for layer in &job_desc.selected_layers {
3315 948 : let desc = layer.layer_desc();
3316 948 : if desc.is_delta() {
3317 528 : stat.visit_delta_layer(desc.file_size());
3318 528 : } else {
3319 420 : stat.visit_image_layer(desc.file_size());
3320 420 : }
3321 : }
3322 :
3323 : // Step 1: construct a k-merge iterator over all layers.
3324 : // Also, verify if the layer map can be split by drawing a horizontal line at every LSN start/end split point.
3325 324 : let layer_names = job_desc
3326 324 : .selected_layers
3327 324 : .iter()
3328 948 : .map(|layer| layer.layer_desc().layer_name())
3329 324 : .collect_vec();
3330 324 : if let Some(err) = check_valid_layermap(&layer_names) {
3331 0 : return Err(CompactionError::Other(anyhow!(
3332 0 : "gc-compaction layer map check failed because {}, cannot proceed with compaction due to potential data loss",
3333 0 : err
3334 0 : )));
3335 324 : }
3336 324 : // The maximum LSN we are processing in this compaction loop
3337 324 : let end_lsn = job_desc
3338 324 : .selected_layers
3339 324 : .iter()
3340 948 : .map(|l| l.layer_desc().lsn_range.end)
3341 324 : .max()
3342 324 : .unwrap();
3343 324 : let mut delta_layers = Vec::new();
3344 324 : let mut image_layers = Vec::new();
3345 324 : let mut downloaded_layers = Vec::new();
3346 324 : let mut total_downloaded_size = 0;
3347 324 : let mut total_layer_size = 0;
3348 1272 : for layer in &job_desc.selected_layers {
3349 948 : if layer
3350 948 : .needs_download()
3351 948 : .await
3352 948 : .context("failed to check if layer needs download")
3353 948 : .map_err(CompactionError::Other)?
3354 948 : .is_some()
3355 0 : {
3356 0 : total_downloaded_size += layer.layer_desc().file_size;
3357 948 : }
3358 948 : total_layer_size += layer.layer_desc().file_size;
3359 948 : if cancel.is_cancelled() {
3360 0 : return Err(CompactionError::ShuttingDown);
3361 948 : }
3362 948 : let should_yield = yield_for_l0
3363 0 : && self
3364 0 : .l0_compaction_trigger
3365 0 : .notified()
3366 0 : .now_or_never()
3367 0 : .is_some();
3368 948 : if should_yield {
3369 0 : tracing::info!("preempt gc-compaction when downloading layers: too many L0 layers");
3370 0 : return Ok(CompactionOutcome::YieldForL0);
3371 948 : }
3372 948 : let resident_layer = layer
3373 948 : .download_and_keep_resident(ctx)
3374 948 : .await
3375 948 : .context("failed to download and keep resident layer")
3376 948 : .map_err(CompactionError::Other)?;
3377 948 : downloaded_layers.push(resident_layer);
3378 : }
3379 324 : info!(
3380 0 : "finish downloading layers, downloaded={}, total={}, ratio={:.2}",
3381 0 : total_downloaded_size,
3382 0 : total_layer_size,
3383 0 : total_downloaded_size as f64 / total_layer_size as f64
3384 : );
3385 1272 : for resident_layer in &downloaded_layers {
3386 948 : if resident_layer.layer_desc().is_delta() {
3387 528 : let layer = resident_layer
3388 528 : .get_as_delta(ctx)
3389 528 : .await
3390 528 : .context("failed to get delta layer")
3391 528 : .map_err(CompactionError::Other)?;
3392 528 : delta_layers.push(layer);
3393 : } else {
3394 420 : let layer = resident_layer
3395 420 : .get_as_image(ctx)
3396 420 : .await
3397 420 : .context("failed to get image layer")
3398 420 : .map_err(CompactionError::Other)?;
3399 420 : image_layers.push(layer);
3400 : }
3401 : }
3402 324 : let (dense_ks, sparse_ks) = self
3403 324 : .collect_gc_compaction_keyspace()
3404 324 : .await
3405 324 : .context("failed to collect gc compaction keyspace")
3406 324 : .map_err(CompactionError::Other)?;
3407 324 : let mut merge_iter = FilterIterator::create(
3408 324 : MergeIterator::create_with_options(
3409 324 : &delta_layers,
3410 324 : &image_layers,
3411 324 : ctx,
3412 324 : 128 * 8192, /* 1MB buffer for each of the inner iterators */
3413 324 : 128,
3414 324 : ),
3415 324 : dense_ks,
3416 324 : sparse_ks,
3417 324 : )
3418 324 : .context("failed to create filter iterator")
3419 324 : .map_err(CompactionError::Other)?;
3420 :
3421 324 : let time_download_layer = timer.elapsed();
3422 324 : let mut timer = Instant::now();
3423 324 :
3424 324 : // Step 2: Produce images+deltas.
3425 324 : let mut accumulated_values = Vec::new();
3426 324 : let mut last_key: Option<Key> = None;
3427 :
3428 : // Only create image layers when there is no ancestor branches. TODO: create covering image layer
3429 : // when some condition meet.
3430 324 : let mut image_layer_writer = if !has_data_below {
3431 : Some(
3432 264 : SplitImageLayerWriter::new(
3433 264 : self.conf,
3434 264 : self.timeline_id,
3435 264 : self.tenant_shard_id,
3436 264 : job_desc.compaction_key_range.start,
3437 264 : lowest_retain_lsn,
3438 264 : self.get_compaction_target_size(),
3439 264 : &self.gate,
3440 264 : self.cancel.clone(),
3441 264 : ctx,
3442 264 : )
3443 264 : .await
3444 264 : .context("failed to create image layer writer")
3445 264 : .map_err(CompactionError::Other)?,
3446 : )
3447 : } else {
3448 60 : None
3449 : };
3450 :
3451 324 : let mut delta_layer_writer = SplitDeltaLayerWriter::new(
3452 324 : self.conf,
3453 324 : self.timeline_id,
3454 324 : self.tenant_shard_id,
3455 324 : lowest_retain_lsn..end_lsn,
3456 324 : self.get_compaction_target_size(),
3457 324 : &self.gate,
3458 324 : self.cancel.clone(),
3459 324 : )
3460 324 : .await
3461 324 : .context("failed to create delta layer writer")
3462 324 : .map_err(CompactionError::Other)?;
3463 :
3464 : #[derive(Default)]
3465 : struct RewritingLayers {
3466 : before: Option<DeltaLayerWriter>,
3467 : after: Option<DeltaLayerWriter>,
3468 : }
3469 324 : let mut delta_layer_rewriters = HashMap::<Arc<PersistentLayerKey>, RewritingLayers>::new();
3470 :
3471 : /// When compacting not at a bottom range (=`[0,X)`) of the root branch, we "have data below" (`has_data_below=true`).
3472 : /// The two cases are compaction in ancestor branches and when `compact_lsn_range.start` is set.
3473 : /// In those cases, we need to pull up data from below the LSN range we're compaction.
3474 : ///
3475 : /// This function unifies the cases so that later code doesn't have to think about it.
3476 : ///
3477 : /// Currently, we always get the ancestor image for each key in the child branch no matter whether the image
3478 : /// is needed for reconstruction. This should be fixed in the future.
3479 : ///
3480 : /// Furthermore, we should do vectored get instead of a single get, or better, use k-merge for ancestor
3481 : /// images.
3482 3840 : async fn get_ancestor_image(
3483 3840 : this_tline: &Arc<Timeline>,
3484 3840 : key: Key,
3485 3840 : ctx: &RequestContext,
3486 3840 : has_data_below: bool,
3487 3840 : history_lsn_point: Lsn,
3488 3840 : ) -> anyhow::Result<Option<(Key, Lsn, Bytes)>> {
3489 3840 : if !has_data_below {
3490 3612 : return Ok(None);
3491 228 : };
3492 : // This function is implemented as a get of the current timeline at ancestor LSN, therefore reusing
3493 : // as much existing code as possible.
3494 228 : let img = this_tline.get(key, history_lsn_point, ctx).await?;
3495 228 : Ok(Some((key, history_lsn_point, img)))
3496 3840 : }
3497 :
3498 : // Actually, we can decide not to write to the image layer at all at this point because
3499 : // the key and LSN range are determined. However, to keep things simple here, we still
3500 : // create this writer, and discard the writer in the end.
3501 324 : let mut time_to_first_kv_pair = None;
3502 :
3503 5952 : while let Some(((key, lsn, val), desc)) = merge_iter
3504 5952 : .next_with_trace()
3505 5952 : .await
3506 5952 : .context("failed to get next key-value pair")
3507 5952 : .map_err(CompactionError::Other)?
3508 : {
3509 5640 : if time_to_first_kv_pair.is_none() {
3510 324 : time_to_first_kv_pair = Some(timer.elapsed());
3511 324 : timer = Instant::now();
3512 5316 : }
3513 :
3514 5640 : if cancel.is_cancelled() {
3515 0 : return Err(CompactionError::ShuttingDown);
3516 5640 : }
3517 :
3518 5640 : let should_yield = yield_for_l0
3519 0 : && self
3520 0 : .l0_compaction_trigger
3521 0 : .notified()
3522 0 : .now_or_never()
3523 0 : .is_some();
3524 5640 : if should_yield {
3525 0 : tracing::info!("preempt gc-compaction in the main loop: too many L0 layers");
3526 0 : return Ok(CompactionOutcome::YieldForL0);
3527 5640 : }
3528 5640 : if self.shard_identity.is_key_disposable(&key) {
3529 : // If this shard does not need to store this key, simply skip it.
3530 : //
3531 : // This is not handled in the filter iterator because shard is determined by hash.
3532 : // Therefore, it does not give us any performance benefit to do things like skip
3533 : // a whole layer file as handling key spaces (ranges).
3534 0 : if cfg!(debug_assertions) {
3535 0 : let shard = self.shard_identity.shard_index();
3536 0 : let owner = self.shard_identity.get_shard_number(&key);
3537 0 : panic!("key {key} does not belong on shard {shard}, owned by {owner}");
3538 0 : }
3539 0 : continue;
3540 5640 : }
3541 5640 : if !job_desc.compaction_key_range.contains(&key) {
3542 384 : if !desc.is_delta {
3543 360 : continue;
3544 24 : }
3545 24 : let rewriter = delta_layer_rewriters.entry(desc.clone()).or_default();
3546 24 : let rewriter = if key < job_desc.compaction_key_range.start {
3547 0 : if rewriter.before.is_none() {
3548 0 : rewriter.before = Some(
3549 0 : DeltaLayerWriter::new(
3550 0 : self.conf,
3551 0 : self.timeline_id,
3552 0 : self.tenant_shard_id,
3553 0 : desc.key_range.start,
3554 0 : desc.lsn_range.clone(),
3555 0 : &self.gate,
3556 0 : self.cancel.clone(),
3557 0 : ctx,
3558 0 : )
3559 0 : .await
3560 0 : .context("failed to create delta layer writer")
3561 0 : .map_err(CompactionError::Other)?,
3562 : );
3563 0 : }
3564 0 : rewriter.before.as_mut().unwrap()
3565 24 : } else if key >= job_desc.compaction_key_range.end {
3566 24 : if rewriter.after.is_none() {
3567 12 : rewriter.after = Some(
3568 12 : DeltaLayerWriter::new(
3569 12 : self.conf,
3570 12 : self.timeline_id,
3571 12 : self.tenant_shard_id,
3572 12 : job_desc.compaction_key_range.end,
3573 12 : desc.lsn_range.clone(),
3574 12 : &self.gate,
3575 12 : self.cancel.clone(),
3576 12 : ctx,
3577 12 : )
3578 12 : .await
3579 12 : .context("failed to create delta layer writer")
3580 12 : .map_err(CompactionError::Other)?,
3581 : );
3582 12 : }
3583 24 : rewriter.after.as_mut().unwrap()
3584 : } else {
3585 0 : unreachable!()
3586 : };
3587 24 : rewriter
3588 24 : .put_value(key, lsn, val, ctx)
3589 24 : .await
3590 24 : .context("failed to put value")
3591 24 : .map_err(CompactionError::Other)?;
3592 24 : continue;
3593 5256 : }
3594 5256 : match val {
3595 3780 : Value::Image(_) => stat.visit_image_key(&val),
3596 1476 : Value::WalRecord(_) => stat.visit_wal_key(&val),
3597 : }
3598 5256 : if last_key.is_none() || last_key.as_ref() == Some(&key) {
3599 1728 : if last_key.is_none() {
3600 324 : last_key = Some(key);
3601 1404 : }
3602 1728 : accumulated_values.push((key, lsn, val));
3603 : } else {
3604 3528 : let last_key: &mut Key = last_key.as_mut().unwrap();
3605 3528 : stat.on_unique_key_visited(); // TODO: adjust statistics for partial compaction
3606 3528 : let retention = self
3607 3528 : .generate_key_retention(
3608 3528 : *last_key,
3609 3528 : &accumulated_values,
3610 3528 : job_desc.gc_cutoff,
3611 3528 : &job_desc.retain_lsns_below_horizon,
3612 3528 : COMPACTION_DELTA_THRESHOLD,
3613 3528 : get_ancestor_image(self, *last_key, ctx, has_data_below, lowest_retain_lsn)
3614 3528 : .await
3615 3528 : .context("failed to get ancestor image")
3616 3528 : .map_err(CompactionError::Other)?,
3617 3528 : verification,
3618 3528 : )
3619 3528 : .await
3620 3528 : .context("failed to generate key retention")
3621 3528 : .map_err(CompactionError::Other)?;
3622 3516 : retention
3623 3516 : .pipe_to(
3624 3516 : *last_key,
3625 3516 : &mut delta_layer_writer,
3626 3516 : image_layer_writer.as_mut(),
3627 3516 : &mut stat,
3628 3516 : ctx,
3629 3516 : )
3630 3516 : .await
3631 3516 : .context("failed to pipe to delta layer writer")
3632 3516 : .map_err(CompactionError::Other)?;
3633 3516 : accumulated_values.clear();
3634 3516 : *last_key = key;
3635 3516 : accumulated_values.push((key, lsn, val));
3636 : }
3637 : }
3638 :
3639 : // TODO: move the below part to the loop body
3640 312 : let Some(last_key) = last_key else {
3641 0 : return Err(CompactionError::Other(anyhow!(
3642 0 : "no keys produced during compaction"
3643 0 : )));
3644 : };
3645 312 : stat.on_unique_key_visited();
3646 :
3647 312 : let retention = self
3648 312 : .generate_key_retention(
3649 312 : last_key,
3650 312 : &accumulated_values,
3651 312 : job_desc.gc_cutoff,
3652 312 : &job_desc.retain_lsns_below_horizon,
3653 312 : COMPACTION_DELTA_THRESHOLD,
3654 312 : get_ancestor_image(self, last_key, ctx, has_data_below, lowest_retain_lsn)
3655 312 : .await
3656 312 : .context("failed to get ancestor image")
3657 312 : .map_err(CompactionError::Other)?,
3658 312 : verification,
3659 312 : )
3660 312 : .await
3661 312 : .context("failed to generate key retention")
3662 312 : .map_err(CompactionError::Other)?;
3663 312 : retention
3664 312 : .pipe_to(
3665 312 : last_key,
3666 312 : &mut delta_layer_writer,
3667 312 : image_layer_writer.as_mut(),
3668 312 : &mut stat,
3669 312 : ctx,
3670 312 : )
3671 312 : .await
3672 312 : .context("failed to pipe to delta layer writer")
3673 312 : .map_err(CompactionError::Other)?;
3674 : // end: move the above part to the loop body
3675 :
3676 312 : let time_main_loop = timer.elapsed();
3677 312 : let timer = Instant::now();
3678 312 :
3679 312 : let mut rewrote_delta_layers = Vec::new();
3680 324 : for (key, writers) in delta_layer_rewriters {
3681 12 : if let Some(delta_writer_before) = writers.before {
3682 0 : let (desc, path) = delta_writer_before
3683 0 : .finish(job_desc.compaction_key_range.start, ctx)
3684 0 : .await
3685 0 : .context("failed to finish delta layer writer")
3686 0 : .map_err(CompactionError::Other)?;
3687 0 : let layer = Layer::finish_creating(self.conf, self, desc, &path)
3688 0 : .context("failed to finish creating delta layer")
3689 0 : .map_err(CompactionError::Other)?;
3690 0 : rewrote_delta_layers.push(layer);
3691 12 : }
3692 12 : if let Some(delta_writer_after) = writers.after {
3693 12 : let (desc, path) = delta_writer_after
3694 12 : .finish(key.key_range.end, ctx)
3695 12 : .await
3696 12 : .context("failed to finish delta layer writer")
3697 12 : .map_err(CompactionError::Other)?;
3698 12 : let layer = Layer::finish_creating(self.conf, self, desc, &path)
3699 12 : .context("failed to finish creating delta layer")
3700 12 : .map_err(CompactionError::Other)?;
3701 12 : rewrote_delta_layers.push(layer);
3702 0 : }
3703 : }
3704 :
3705 444 : let discard = |key: &PersistentLayerKey| {
3706 444 : let key = key.clone();
3707 444 : async move { KeyHistoryRetention::discard_key(&key, self, dry_run).await }
3708 444 : };
3709 :
3710 312 : let produced_image_layers = if let Some(writer) = image_layer_writer {
3711 252 : if !dry_run {
3712 228 : let end_key = job_desc.compaction_key_range.end;
3713 228 : writer
3714 228 : .finish_with_discard_fn(self, ctx, end_key, discard)
3715 228 : .await
3716 228 : .context("failed to finish image layer writer")
3717 228 : .map_err(CompactionError::Other)?
3718 : } else {
3719 24 : drop(writer);
3720 24 : Vec::new()
3721 : }
3722 : } else {
3723 60 : Vec::new()
3724 : };
3725 :
3726 312 : let produced_delta_layers = if !dry_run {
3727 288 : delta_layer_writer
3728 288 : .finish_with_discard_fn(self, ctx, discard)
3729 288 : .await
3730 288 : .context("failed to finish delta layer writer")
3731 288 : .map_err(CompactionError::Other)?
3732 : } else {
3733 24 : drop(delta_layer_writer);
3734 24 : Vec::new()
3735 : };
3736 :
3737 : // TODO: make image/delta/rewrote_delta layers generation atomic. At this point, we already generated resident layers, and if
3738 : // compaction is cancelled at this point, we might have some layers that are not cleaned up.
3739 312 : let mut compact_to = Vec::new();
3740 312 : let mut keep_layers = HashSet::new();
3741 312 : let produced_delta_layers_len = produced_delta_layers.len();
3742 312 : let produced_image_layers_len = produced_image_layers.len();
3743 312 :
3744 312 : let layer_selection_by_key = job_desc
3745 312 : .selected_layers
3746 312 : .iter()
3747 912 : .map(|l| (l.layer_desc().key(), l.layer_desc().clone()))
3748 312 : .collect::<HashMap<_, _>>();
3749 :
3750 528 : for action in produced_delta_layers {
3751 216 : match action {
3752 132 : BatchWriterResult::Produced(layer) => {
3753 132 : if cfg!(debug_assertions) {
3754 132 : info!("produced delta layer: {}", layer.layer_desc().key());
3755 0 : }
3756 132 : stat.produce_delta_layer(layer.layer_desc().file_size());
3757 132 : compact_to.push(layer);
3758 : }
3759 84 : BatchWriterResult::Discarded(l) => {
3760 84 : if cfg!(debug_assertions) {
3761 84 : info!("discarded delta layer: {}", l);
3762 0 : }
3763 84 : if let Some(layer_desc) = layer_selection_by_key.get(&l) {
3764 84 : stat.discard_delta_layer(layer_desc.file_size());
3765 84 : } else {
3766 0 : tracing::warn!(
3767 0 : "discarded delta layer not in layer_selection: {}, produced a layer outside of the compaction key range?",
3768 : l
3769 : );
3770 0 : stat.discard_delta_layer(0);
3771 : }
3772 84 : keep_layers.insert(l);
3773 : }
3774 : }
3775 : }
3776 324 : for layer in &rewrote_delta_layers {
3777 12 : debug!(
3778 0 : "produced rewritten delta layer: {}",
3779 0 : layer.layer_desc().key()
3780 : );
3781 : // For now, we include rewritten delta layer size in the "produce_delta_layer". We could
3782 : // make it a separate statistics in the future.
3783 12 : stat.produce_delta_layer(layer.layer_desc().file_size());
3784 : }
3785 312 : compact_to.extend(rewrote_delta_layers);
3786 540 : for action in produced_image_layers {
3787 228 : match action {
3788 180 : BatchWriterResult::Produced(layer) => {
3789 180 : debug!("produced image layer: {}", layer.layer_desc().key());
3790 180 : stat.produce_image_layer(layer.layer_desc().file_size());
3791 180 : compact_to.push(layer);
3792 : }
3793 48 : BatchWriterResult::Discarded(l) => {
3794 48 : debug!("discarded image layer: {}", l);
3795 48 : if let Some(layer_desc) = layer_selection_by_key.get(&l) {
3796 48 : stat.discard_image_layer(layer_desc.file_size());
3797 48 : } else {
3798 0 : tracing::warn!(
3799 0 : "discarded image layer not in layer_selection: {}, produced a layer outside of the compaction key range?",
3800 : l
3801 : );
3802 0 : stat.discard_image_layer(0);
3803 : }
3804 48 : keep_layers.insert(l);
3805 : }
3806 : }
3807 : }
3808 :
3809 312 : let mut layer_selection = job_desc.selected_layers;
3810 :
3811 : // Partial compaction might select more data than it processes, e.g., if
3812 : // the compaction_key_range only partially overlaps:
3813 : //
3814 : // [---compaction_key_range---]
3815 : // [---A----][----B----][----C----][----D----]
3816 : //
3817 : // For delta layers, we will rewrite the layers so that it is cut exactly at
3818 : // the compaction key range, so we can always discard them. However, for image
3819 : // layers, as we do not rewrite them for now, we need to handle them differently.
3820 : // Assume image layers A, B, C, D are all in the `layer_selection`.
3821 : //
3822 : // The created image layers contain whatever is needed from B, C, and from
3823 : // `----]` of A, and from `[---` of D.
3824 : //
3825 : // In contrast, `[---A` and `D----]` have not been processed, so, we must
3826 : // keep that data.
3827 : //
3828 : // The solution for now is to keep A and D completely if they are image layers.
3829 : // (layer_selection is what we'll remove from the layer map, so, retain what
3830 : // is _not_ fully covered by compaction_key_range).
3831 1224 : for layer in &layer_selection {
3832 912 : if !layer.layer_desc().is_delta() {
3833 396 : if !overlaps_with(
3834 396 : &layer.layer_desc().key_range,
3835 396 : &job_desc.compaction_key_range,
3836 396 : ) {
3837 0 : return Err(CompactionError::Other(anyhow!(
3838 0 : "violated constraint: image layer outside of compaction key range"
3839 0 : )));
3840 396 : }
3841 396 : if !fully_contains(
3842 396 : &job_desc.compaction_key_range,
3843 396 : &layer.layer_desc().key_range,
3844 396 : ) {
3845 48 : keep_layers.insert(layer.layer_desc().key());
3846 348 : }
3847 516 : }
3848 : }
3849 :
3850 912 : layer_selection.retain(|x| !keep_layers.contains(&x.layer_desc().key()));
3851 312 :
3852 312 : let time_final_phase = timer.elapsed();
3853 312 :
3854 312 : stat.time_final_phase_secs = time_final_phase.as_secs_f64();
3855 312 : stat.time_to_first_kv_pair_secs = time_to_first_kv_pair
3856 312 : .unwrap_or(Duration::ZERO)
3857 312 : .as_secs_f64();
3858 312 : stat.time_main_loop_secs = time_main_loop.as_secs_f64();
3859 312 : stat.time_acquire_lock_secs = time_acquire_lock.as_secs_f64();
3860 312 : stat.time_download_layer_secs = time_download_layer.as_secs_f64();
3861 312 : stat.time_analyze_secs = time_analyze.as_secs_f64();
3862 312 : stat.time_total_secs = begin_timer.elapsed().as_secs_f64();
3863 312 : stat.finalize();
3864 312 :
3865 312 : info!(
3866 0 : "gc-compaction statistics: {}",
3867 0 : serde_json::to_string(&stat)
3868 0 : .context("failed to serialize gc-compaction statistics")
3869 0 : .map_err(CompactionError::Other)?
3870 : );
3871 :
3872 312 : if dry_run {
3873 24 : return Ok(CompactionOutcome::Done);
3874 288 : }
3875 288 :
3876 288 : info!(
3877 0 : "produced {} delta layers and {} image layers, {} layers are kept",
3878 0 : produced_delta_layers_len,
3879 0 : produced_image_layers_len,
3880 0 : keep_layers.len()
3881 : );
3882 :
3883 : // Step 3: Place back to the layer map.
3884 :
3885 : // First, do a sanity check to ensure the newly-created layer map does not contain overlaps.
3886 288 : let all_layers = {
3887 288 : let guard = self.layers.read().await;
3888 288 : let layer_map = guard.layer_map()?;
3889 288 : layer_map.iter_historic_layers().collect_vec()
3890 288 : };
3891 288 :
3892 288 : let mut final_layers = all_layers
3893 288 : .iter()
3894 1284 : .map(|layer| layer.layer_name())
3895 288 : .collect::<HashSet<_>>();
3896 912 : for layer in &layer_selection {
3897 624 : final_layers.remove(&layer.layer_desc().layer_name());
3898 624 : }
3899 612 : for layer in &compact_to {
3900 324 : final_layers.insert(layer.layer_desc().layer_name());
3901 324 : }
3902 288 : let final_layers = final_layers.into_iter().collect_vec();
3903 :
3904 : // TODO: move this check before we call `finish` on image layer writers. However, this will require us to get the layer name before we finish
3905 : // the writer, so potentially, we will need a function like `ImageLayerBatchWriter::get_all_pending_layer_keys` to get all the keys that are
3906 : // in the writer before finalizing the persistent layers. Now we would leave some dangling layers on the disk if the check fails.
3907 288 : if let Some(err) = check_valid_layermap(&final_layers) {
3908 0 : return Err(CompactionError::Other(anyhow!(
3909 0 : "gc-compaction layer map check failed after compaction because {}, compaction result not applied to the layer map due to potential data loss",
3910 0 : err
3911 0 : )));
3912 288 : }
3913 :
3914 : // Between the sanity check and this compaction update, there could be new layers being flushed, but it should be fine because we only
3915 : // operate on L1 layers.
3916 : {
3917 : // Gc-compaction will rewrite the history of a key. This could happen in two ways:
3918 : //
3919 : // 1. We create an image layer to replace all the deltas below the compact LSN. In this case, assume
3920 : // we have 2 delta layers A and B, both below the compact LSN. We create an image layer I to replace
3921 : // A and B at the compact LSN. If the read path finishes reading A, yields, and now we update the layer
3922 : // map, the read path then cannot find any keys below A, reporting a missing key error, while the key
3923 : // now gets stored in I at the compact LSN.
3924 : //
3925 : // --------------- ---------------
3926 : // delta1@LSN20 image1@LSN20
3927 : // --------------- (read path collects delta@LSN20, => --------------- (read path cannot find anything
3928 : // delta1@LSN10 yields) below LSN 20)
3929 : // ---------------
3930 : //
3931 : // 2. We create a delta layer to replace all the deltas below the compact LSN, and in the delta layers,
3932 : // we combines the history of a key into a single image. For example, we have deltas at LSN 1, 2, 3, 4,
3933 : // Assume one delta layer contains LSN 1, 2, 3 and the other contains LSN 4.
3934 : //
3935 : // We let gc-compaction combine delta 2, 3, 4 into an image at LSN 4, which produces a delta layer that
3936 : // contains the delta at LSN 1, the image at LSN 4. If the read path finishes reading the original delta
3937 : // layer containing 4, yields, and we update the layer map to put the delta layer.
3938 : //
3939 : // --------------- ---------------
3940 : // delta1@LSN4 image1@LSN4
3941 : // --------------- (read path collects delta@LSN4, => --------------- (read path collects LSN4 and LSN1,
3942 : // delta1@LSN1-3 yields) delta1@LSN1 which is an invalid history)
3943 : // --------------- ---------------
3944 : //
3945 : // Therefore, the gc-compaction layer update operation should wait for all ongoing reads, block all pending reads,
3946 : // and only allow reads to continue after the update is finished.
3947 :
3948 288 : let update_guard = self.gc_compaction_layer_update_lock.write().await;
3949 : // Acquiring the update guard ensures current read operations end and new read operations are blocked.
3950 : // TODO: can we use `latest_gc_cutoff` Rcu to achieve the same effect?
3951 288 : let mut guard = self.layers.write().await;
3952 288 : guard
3953 288 : .open_mut()?
3954 288 : .finish_gc_compaction(&layer_selection, &compact_to, &self.metrics);
3955 288 : drop(update_guard); // Allow new reads to start ONLY after we finished updating the layer map.
3956 288 : };
3957 288 :
3958 288 : // Schedule an index-only upload to update the `latest_gc_cutoff` in the index_part.json.
3959 288 : // Otherwise, after restart, the index_part only contains the old `latest_gc_cutoff` and
3960 288 : // find_gc_cutoffs will try accessing things below the cutoff. TODO: ideally, this should
3961 288 : // be batched into `schedule_compaction_update`.
3962 288 : let disk_consistent_lsn = self.disk_consistent_lsn.load();
3963 288 : self.schedule_uploads(disk_consistent_lsn, None)
3964 288 : .context("failed to schedule uploads")
3965 288 : .map_err(CompactionError::Other)?;
3966 : // If a layer gets rewritten throughout gc-compaction, we need to keep that layer only in `compact_to` instead
3967 : // of `compact_from`.
3968 288 : let compact_from = {
3969 288 : let mut compact_from = Vec::new();
3970 288 : let mut compact_to_set = HashMap::new();
3971 612 : for layer in &compact_to {
3972 324 : compact_to_set.insert(layer.layer_desc().key(), layer);
3973 324 : }
3974 912 : for layer in &layer_selection {
3975 624 : if let Some(to) = compact_to_set.get(&layer.layer_desc().key()) {
3976 0 : tracing::info!(
3977 0 : "skipping delete {} because found same layer key at different generation {}",
3978 : layer,
3979 : to
3980 : );
3981 624 : } else {
3982 624 : compact_from.push(layer.clone());
3983 624 : }
3984 : }
3985 288 : compact_from
3986 288 : };
3987 288 : self.remote_client
3988 288 : .schedule_compaction_update(&compact_from, &compact_to)?;
3989 :
3990 288 : drop(gc_lock);
3991 288 :
3992 288 : Ok(CompactionOutcome::Done)
3993 336 : }
3994 : }
3995 :
3996 : struct TimelineAdaptor {
3997 : timeline: Arc<Timeline>,
3998 :
3999 : keyspace: (Lsn, KeySpace),
4000 :
4001 : new_deltas: Vec<ResidentLayer>,
4002 : new_images: Vec<ResidentLayer>,
4003 : layers_to_delete: Vec<Arc<PersistentLayerDesc>>,
4004 : }
4005 :
4006 : impl TimelineAdaptor {
4007 0 : pub fn new(timeline: &Arc<Timeline>, keyspace: (Lsn, KeySpace)) -> Self {
4008 0 : Self {
4009 0 : timeline: timeline.clone(),
4010 0 : keyspace,
4011 0 : new_images: Vec::new(),
4012 0 : new_deltas: Vec::new(),
4013 0 : layers_to_delete: Vec::new(),
4014 0 : }
4015 0 : }
4016 :
4017 0 : pub async fn flush_updates(&mut self) -> Result<(), CompactionError> {
4018 0 : let layers_to_delete = {
4019 0 : let guard = self.timeline.layers.read().await;
4020 0 : self.layers_to_delete
4021 0 : .iter()
4022 0 : .map(|x| guard.get_from_desc(x))
4023 0 : .collect::<Vec<Layer>>()
4024 0 : };
4025 0 : self.timeline
4026 0 : .finish_compact_batch(&self.new_deltas, &self.new_images, &layers_to_delete)
4027 0 : .await?;
4028 :
4029 0 : self.timeline
4030 0 : .upload_new_image_layers(std::mem::take(&mut self.new_images))?;
4031 :
4032 0 : self.new_deltas.clear();
4033 0 : self.layers_to_delete.clear();
4034 0 : Ok(())
4035 0 : }
4036 : }
4037 :
4038 : #[derive(Clone)]
4039 : struct ResidentDeltaLayer(ResidentLayer);
4040 : #[derive(Clone)]
4041 : struct ResidentImageLayer(ResidentLayer);
4042 :
4043 : impl CompactionJobExecutor for TimelineAdaptor {
4044 : type Key = pageserver_api::key::Key;
4045 :
4046 : type Layer = OwnArc<PersistentLayerDesc>;
4047 : type DeltaLayer = ResidentDeltaLayer;
4048 : type ImageLayer = ResidentImageLayer;
4049 :
4050 : type RequestContext = crate::context::RequestContext;
4051 :
4052 0 : fn get_shard_identity(&self) -> &ShardIdentity {
4053 0 : self.timeline.get_shard_identity()
4054 0 : }
4055 :
4056 0 : async fn get_layers(
4057 0 : &mut self,
4058 0 : key_range: &Range<Key>,
4059 0 : lsn_range: &Range<Lsn>,
4060 0 : _ctx: &RequestContext,
4061 0 : ) -> anyhow::Result<Vec<OwnArc<PersistentLayerDesc>>> {
4062 0 : self.flush_updates().await?;
4063 :
4064 0 : let guard = self.timeline.layers.read().await;
4065 0 : let layer_map = guard.layer_map()?;
4066 :
4067 0 : let result = layer_map
4068 0 : .iter_historic_layers()
4069 0 : .filter(|l| {
4070 0 : overlaps_with(&l.lsn_range, lsn_range) && overlaps_with(&l.key_range, key_range)
4071 0 : })
4072 0 : .map(OwnArc)
4073 0 : .collect();
4074 0 : Ok(result)
4075 0 : }
4076 :
4077 0 : async fn get_keyspace(
4078 0 : &mut self,
4079 0 : key_range: &Range<Key>,
4080 0 : lsn: Lsn,
4081 0 : _ctx: &RequestContext,
4082 0 : ) -> anyhow::Result<Vec<Range<Key>>> {
4083 0 : if lsn == self.keyspace.0 {
4084 0 : Ok(pageserver_compaction::helpers::intersect_keyspace(
4085 0 : &self.keyspace.1.ranges,
4086 0 : key_range,
4087 0 : ))
4088 : } else {
4089 : // The current compaction implementation only ever requests the key space
4090 : // at the compaction end LSN.
4091 0 : anyhow::bail!("keyspace not available for requested lsn");
4092 : }
4093 0 : }
4094 :
4095 0 : async fn downcast_delta_layer(
4096 0 : &self,
4097 0 : layer: &OwnArc<PersistentLayerDesc>,
4098 0 : ctx: &RequestContext,
4099 0 : ) -> anyhow::Result<Option<ResidentDeltaLayer>> {
4100 0 : // this is a lot more complex than a simple downcast...
4101 0 : if layer.is_delta() {
4102 0 : let l = {
4103 0 : let guard = self.timeline.layers.read().await;
4104 0 : guard.get_from_desc(layer)
4105 : };
4106 0 : let result = l.download_and_keep_resident(ctx).await?;
4107 :
4108 0 : Ok(Some(ResidentDeltaLayer(result)))
4109 : } else {
4110 0 : Ok(None)
4111 : }
4112 0 : }
4113 :
4114 0 : async fn create_image(
4115 0 : &mut self,
4116 0 : lsn: Lsn,
4117 0 : key_range: &Range<Key>,
4118 0 : ctx: &RequestContext,
4119 0 : ) -> anyhow::Result<()> {
4120 0 : Ok(self.create_image_impl(lsn, key_range, ctx).await?)
4121 0 : }
4122 :
4123 0 : async fn create_delta(
4124 0 : &mut self,
4125 0 : lsn_range: &Range<Lsn>,
4126 0 : key_range: &Range<Key>,
4127 0 : input_layers: &[ResidentDeltaLayer],
4128 0 : ctx: &RequestContext,
4129 0 : ) -> anyhow::Result<()> {
4130 0 : debug!("Create new layer {}..{}", lsn_range.start, lsn_range.end);
4131 :
4132 0 : let mut all_entries = Vec::new();
4133 0 : for dl in input_layers.iter() {
4134 0 : all_entries.extend(dl.load_keys(ctx).await?);
4135 : }
4136 :
4137 : // The current stdlib sorting implementation is designed in a way where it is
4138 : // particularly fast where the slice is made up of sorted sub-ranges.
4139 0 : all_entries.sort_by_key(|DeltaEntry { key, lsn, .. }| (*key, *lsn));
4140 :
4141 0 : let mut writer = DeltaLayerWriter::new(
4142 0 : self.timeline.conf,
4143 0 : self.timeline.timeline_id,
4144 0 : self.timeline.tenant_shard_id,
4145 0 : key_range.start,
4146 0 : lsn_range.clone(),
4147 0 : &self.timeline.gate,
4148 0 : self.timeline.cancel.clone(),
4149 0 : ctx,
4150 0 : )
4151 0 : .await?;
4152 :
4153 0 : let mut dup_values = 0;
4154 0 :
4155 0 : // This iterator walks through all key-value pairs from all the layers
4156 0 : // we're compacting, in key, LSN order.
4157 0 : let mut prev: Option<(Key, Lsn)> = None;
4158 : for &DeltaEntry {
4159 0 : key, lsn, ref val, ..
4160 0 : } in all_entries.iter()
4161 : {
4162 0 : if prev == Some((key, lsn)) {
4163 : // This is a duplicate. Skip it.
4164 : //
4165 : // It can happen if compaction is interrupted after writing some
4166 : // layers but not all, and we are compacting the range again.
4167 : // The calculations in the algorithm assume that there are no
4168 : // duplicates, so the math on targeted file size is likely off,
4169 : // and we will create smaller files than expected.
4170 0 : dup_values += 1;
4171 0 : continue;
4172 0 : }
4173 :
4174 0 : let value = val.load(ctx).await?;
4175 :
4176 0 : writer.put_value(key, lsn, value, ctx).await?;
4177 :
4178 0 : prev = Some((key, lsn));
4179 : }
4180 :
4181 0 : if dup_values > 0 {
4182 0 : warn!("delta layer created with {} duplicate values", dup_values);
4183 0 : }
4184 :
4185 0 : fail_point!("delta-layer-writer-fail-before-finish", |_| {
4186 0 : Err(anyhow::anyhow!(
4187 0 : "failpoint delta-layer-writer-fail-before-finish"
4188 0 : ))
4189 0 : });
4190 :
4191 0 : let (desc, path) = writer.finish(prev.unwrap().0.next(), ctx).await?;
4192 0 : let new_delta_layer =
4193 0 : Layer::finish_creating(self.timeline.conf, &self.timeline, desc, &path)?;
4194 :
4195 0 : self.new_deltas.push(new_delta_layer);
4196 0 : Ok(())
4197 0 : }
4198 :
4199 0 : async fn delete_layer(
4200 0 : &mut self,
4201 0 : layer: &OwnArc<PersistentLayerDesc>,
4202 0 : _ctx: &RequestContext,
4203 0 : ) -> anyhow::Result<()> {
4204 0 : self.layers_to_delete.push(layer.clone().0);
4205 0 : Ok(())
4206 0 : }
4207 : }
4208 :
4209 : impl TimelineAdaptor {
4210 0 : async fn create_image_impl(
4211 0 : &mut self,
4212 0 : lsn: Lsn,
4213 0 : key_range: &Range<Key>,
4214 0 : ctx: &RequestContext,
4215 0 : ) -> Result<(), CreateImageLayersError> {
4216 0 : let timer = self.timeline.metrics.create_images_time_histo.start_timer();
4217 :
4218 0 : let image_layer_writer = ImageLayerWriter::new(
4219 0 : self.timeline.conf,
4220 0 : self.timeline.timeline_id,
4221 0 : self.timeline.tenant_shard_id,
4222 0 : key_range,
4223 0 : lsn,
4224 0 : &self.timeline.gate,
4225 0 : self.timeline.cancel.clone(),
4226 0 : ctx,
4227 0 : )
4228 0 : .await?;
4229 :
4230 0 : fail_point!("image-layer-writer-fail-before-finish", |_| {
4231 0 : Err(CreateImageLayersError::Other(anyhow::anyhow!(
4232 0 : "failpoint image-layer-writer-fail-before-finish"
4233 0 : )))
4234 0 : });
4235 :
4236 0 : let keyspace = KeySpace {
4237 0 : ranges: self.get_keyspace(key_range, lsn, ctx).await?,
4238 : };
4239 : // TODO set proper (stateful) start. The create_image_layer_for_rel_blocks function mostly
4240 0 : let outcome = self
4241 0 : .timeline
4242 0 : .create_image_layer_for_rel_blocks(
4243 0 : &keyspace,
4244 0 : image_layer_writer,
4245 0 : lsn,
4246 0 : ctx,
4247 0 : key_range.clone(),
4248 0 : IoConcurrency::sequential(),
4249 0 : )
4250 0 : .await?;
4251 :
4252 : if let ImageLayerCreationOutcome::Generated {
4253 0 : unfinished_image_layer,
4254 0 : } = outcome
4255 : {
4256 0 : let (desc, path) = unfinished_image_layer.finish(ctx).await?;
4257 0 : let image_layer =
4258 0 : Layer::finish_creating(self.timeline.conf, &self.timeline, desc, &path)?;
4259 0 : self.new_images.push(image_layer);
4260 0 : }
4261 :
4262 0 : timer.stop_and_record();
4263 0 :
4264 0 : Ok(())
4265 0 : }
4266 : }
4267 :
4268 : impl CompactionRequestContext for crate::context::RequestContext {}
4269 :
4270 : #[derive(Debug, Clone)]
4271 : pub struct OwnArc<T>(pub Arc<T>);
4272 :
4273 : impl<T> Deref for OwnArc<T> {
4274 : type Target = <Arc<T> as Deref>::Target;
4275 0 : fn deref(&self) -> &Self::Target {
4276 0 : &self.0
4277 0 : }
4278 : }
4279 :
4280 : impl<T> AsRef<T> for OwnArc<T> {
4281 0 : fn as_ref(&self) -> &T {
4282 0 : self.0.as_ref()
4283 0 : }
4284 : }
4285 :
4286 : impl CompactionLayer<Key> for OwnArc<PersistentLayerDesc> {
4287 0 : fn key_range(&self) -> &Range<Key> {
4288 0 : &self.key_range
4289 0 : }
4290 0 : fn lsn_range(&self) -> &Range<Lsn> {
4291 0 : &self.lsn_range
4292 0 : }
4293 0 : fn file_size(&self) -> u64 {
4294 0 : self.file_size
4295 0 : }
4296 0 : fn short_id(&self) -> std::string::String {
4297 0 : self.as_ref().short_id().to_string()
4298 0 : }
4299 0 : fn is_delta(&self) -> bool {
4300 0 : self.as_ref().is_delta()
4301 0 : }
4302 : }
4303 :
4304 : impl CompactionLayer<Key> for OwnArc<DeltaLayer> {
4305 0 : fn key_range(&self) -> &Range<Key> {
4306 0 : &self.layer_desc().key_range
4307 0 : }
4308 0 : fn lsn_range(&self) -> &Range<Lsn> {
4309 0 : &self.layer_desc().lsn_range
4310 0 : }
4311 0 : fn file_size(&self) -> u64 {
4312 0 : self.layer_desc().file_size
4313 0 : }
4314 0 : fn short_id(&self) -> std::string::String {
4315 0 : self.layer_desc().short_id().to_string()
4316 0 : }
4317 0 : fn is_delta(&self) -> bool {
4318 0 : true
4319 0 : }
4320 : }
4321 :
4322 : impl CompactionLayer<Key> for ResidentDeltaLayer {
4323 0 : fn key_range(&self) -> &Range<Key> {
4324 0 : &self.0.layer_desc().key_range
4325 0 : }
4326 0 : fn lsn_range(&self) -> &Range<Lsn> {
4327 0 : &self.0.layer_desc().lsn_range
4328 0 : }
4329 0 : fn file_size(&self) -> u64 {
4330 0 : self.0.layer_desc().file_size
4331 0 : }
4332 0 : fn short_id(&self) -> std::string::String {
4333 0 : self.0.layer_desc().short_id().to_string()
4334 0 : }
4335 0 : fn is_delta(&self) -> bool {
4336 0 : true
4337 0 : }
4338 : }
4339 :
4340 : impl CompactionDeltaLayer<TimelineAdaptor> for ResidentDeltaLayer {
4341 : type DeltaEntry<'a> = DeltaEntry<'a>;
4342 :
4343 0 : async fn load_keys(&self, ctx: &RequestContext) -> anyhow::Result<Vec<DeltaEntry<'_>>> {
4344 0 : self.0.get_as_delta(ctx).await?.index_entries(ctx).await
4345 0 : }
4346 : }
4347 :
4348 : impl CompactionLayer<Key> for ResidentImageLayer {
4349 0 : fn key_range(&self) -> &Range<Key> {
4350 0 : &self.0.layer_desc().key_range
4351 0 : }
4352 0 : fn lsn_range(&self) -> &Range<Lsn> {
4353 0 : &self.0.layer_desc().lsn_range
4354 0 : }
4355 0 : fn file_size(&self) -> u64 {
4356 0 : self.0.layer_desc().file_size
4357 0 : }
4358 0 : fn short_id(&self) -> std::string::String {
4359 0 : self.0.layer_desc().short_id().to_string()
4360 0 : }
4361 0 : fn is_delta(&self) -> bool {
4362 0 : false
4363 0 : }
4364 : }
4365 : impl CompactionImageLayer<TimelineAdaptor> for ResidentImageLayer {}
|
