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