fix

Signed-off-by: Alex Chi Z <chi@neon.tech>

control_plane/src/bin/neon_local.rs

@@ -308,7 +308,8 @@ fn handle_init(init_match: &ArgMatches) -> anyhow::Result<LocalEnv> {
 
     let mut env =
         LocalEnv::parse_config(&toml_file).context("Failed to create neon configuration")?;
-    env.init(pg_version)
+    let force = init_match.get_flag("force");
+    env.init(pg_version, force)
         .context("Failed to initialize neon repository")?;
 
     // Initialize pageserver, create initial tenant and timeline.
@@ -1013,6 +1014,13 @@ fn cli() -> Command {
         .help("If set, the node will be a hot replica on the specified timeline")
         .required(false);
 
+    let force_arg = Arg::new("force")
+        .value_parser(value_parser!(bool))
+        .long("force")
+        .action(ArgAction::SetTrue)
+        .help("Force initialization even if the repository is not empty")
+        .required(false);
+
     Command::new("Neon CLI")
         .arg_required_else_help(true)
         .version(GIT_VERSION)
@@ -1028,6 +1036,7 @@ fn cli() -> Command {
                         .value_name("config"),
                 )
                 .arg(pg_version_arg.clone())
+                .arg(force_arg)
         )
         .subcommand(
             Command::new("timeline")

control_plane/src/local_env.rs

@@ -364,7 +364,7 @@ impl LocalEnv {
     //
     // Initialize a new Neon repository
     //
-    pub fn init(&mut self, pg_version: u32) -> anyhow::Result<()> {
+    pub fn init(&mut self, pg_version: u32, force: bool) -> anyhow::Result<()> {
         // check if config already exists
         let base_path = &self.base_data_dir;
         ensure!(
@@ -372,11 +372,29 @@ impl LocalEnv {
             "repository base path is missing"
         );
 
-        ensure!(
-            !base_path.exists(),
-            "directory '{}' already exists. Perhaps already initialized?",
-            base_path.display()
-        );
+        if base_path.exists() {
+            if force {
+                println!("removing all contents of '{}'", base_path.display());
+                // instead of directly calling `remove_dir_all`, we keep the original dir but removing
+                // all contents inside. This helps if the developer symbol links another directory (i.e.,
+                // S3 local SSD) to the `.neon` base directory.
+                for entry in std::fs::read_dir(base_path)? {
+                    let entry = entry?;
+                    let path = entry.path();
+                    if path.is_dir() {
+                        fs::remove_dir_all(&path)?;
+                    } else {
+                        fs::remove_file(&path)?;
+                    }
+                }
+            } else {
+                bail!(
+                    "directory '{}' already exists. Perhaps already initialized? (Hint: use --force to remove all contents)",
+                    base_path.display()
+                );
+            }
+        }
+
         if !self.pg_bin_dir(pg_version)?.join("postgres").exists() {
             bail!(
                 "Can't find postgres binary at {}",
@@ -392,7 +410,7 @@ impl LocalEnv {
             }
         }
 
-        fs::create_dir(base_path)?;
+        fs::create_dir_all(base_path)?;
 
         // Generate keypair for JWT.
         //

pageserver/ctl/src/draw_timeline_dir.rs

@@ -117,7 +117,8 @@ pub fn main() -> Result<()> {
 
         let mut lsn_diff = (lsn_end - lsn_start) as f32;
         let mut fill = Fill::None;
-        let mut margin = 0.05 * lsn_diff; // Height-dependent margin to disambiguate overlapping deltas
+        let mut ymargin = 0.05 * lsn_diff; // Height-dependent margin to disambiguate overlapping deltas
+        let xmargin = 0.05; // Height-dependent margin to disambiguate overlapping deltas
         let mut lsn_offset = 0.0;
 
         // Fill in and thicken rectangle if it's an
@@ -128,7 +129,7 @@ pub fn main() -> Result<()> {
                 num_images += 1;
                 lsn_diff = 0.3;
                 lsn_offset = -lsn_diff / 2.0;
-                margin = 0.05;
+                ymargin = 0.05;
                 fill = Fill::Color(rgb(0, 0, 0));
             }
             Ordering::Greater => panic!("Invalid lsn range {}-{}", lsn_start, lsn_end),
@@ -137,10 +138,10 @@ pub fn main() -> Result<()> {
         println!(
             "    {}",
             rectangle(
-                key_start as f32 + stretch * margin,
-                stretch * (lsn_max as f32 - (lsn_end as f32 - margin - lsn_offset)),
-                key_diff as f32 - stretch * 2.0 * margin,
-                stretch * (lsn_diff - 2.0 * margin)
+                key_start as f32 + stretch * xmargin,
+                stretch * (lsn_max as f32 - (lsn_end as f32 - ymargin - lsn_offset)),
+                key_diff as f32 - stretch * 2.0 * xmargin,
+                stretch * (lsn_diff - 2.0 * ymargin)
             )
             .fill(fill)
             .stroke(Stroke::Color(rgb(0, 0, 0), 0.1))

pageserver/src/metrics.rs

@@ -53,6 +53,33 @@ pub enum StorageTimeOperation {
     CreateTenant,
 }
 
+pub static NUM_TIERS: Lazy<IntGaugeVec> = Lazy::new(|| {
+    register_int_gauge_vec!(
+        "pageserver_storage_tiers_num",
+        "Number of sorted runs",
+        &["tenant_id", "timeline_id"],
+    )
+    .expect("failed to define a metric")
+});
+
+pub static NUM_COMPACTIONS: Lazy<IntGaugeVec> = Lazy::new(|| {
+    register_int_gauge_vec!(
+        "pageserver_storage_compaction_num",
+        "Number of ongoing compactions",
+        &["tenant_id", "timeline_id"],
+    )
+    .expect("failed to define a metric")
+});
+
+pub static STORAGE_PHYSICAL_SIZE: Lazy<IntGaugeVec> = Lazy::new(|| {
+    register_int_gauge_vec!(
+        "pageserver_storage_physical_size_sum",
+        "Physical size of different types of storage files",
+        &["type", "tenant_id", "timeline_id"],
+    )
+    .expect("failed to define a metric")
+});
+
 pub static STORAGE_TIME_SUM_PER_TIMELINE: Lazy<CounterVec> = Lazy::new(|| {
     register_counter_vec!(
         "pageserver_storage_operations_seconds_sum",
@@ -392,6 +419,8 @@ const STORAGE_IO_TIME_OPERATIONS: &[&str] = &[
 
 const STORAGE_IO_SIZE_OPERATIONS: &[&str] = &["read", "write"];
 
+pub const STORAGE_PHYSICAL_SIZE_FILE_TYPE: &[&str] = &["image", "delta", "partial-image"];
+
 pub static STORAGE_IO_TIME: Lazy<HistogramVec> = Lazy::new(|| {
     register_histogram_vec!(
         "pageserver_io_operations_seconds",
@@ -773,6 +802,8 @@ pub struct TimelineMetrics {
     pub persistent_bytes_written: IntCounter,
     pub evictions: IntCounter,
     pub evictions_with_low_residence_duration: std::sync::RwLock<EvictionsWithLowResidenceDuration>,
+    pub num_tiers: IntGauge,
+    pub num_compactions: IntGauge,
 }
 
 impl TimelineMetrics {
@@ -838,6 +869,12 @@ impl TimelineMetrics {
             .unwrap();
         let evictions_with_low_residence_duration =
             evictions_with_low_residence_duration_builder.build(&tenant_id, &timeline_id);
+        let num_tiers = NUM_TIERS
+            .get_metric_with_label_values(&[&tenant_id, &timeline_id])
+            .unwrap();
+        let num_compactions = NUM_COMPACTIONS
+            .get_metric_with_label_values(&[&tenant_id, &timeline_id])
+            .unwrap();
 
         TimelineMetrics {
             tenant_id,
@@ -864,6 +901,8 @@ impl TimelineMetrics {
                 evictions_with_low_residence_duration,
             ),
             read_num_fs_layers,
+            num_tiers,
+            num_compactions,
         }
     }
 }
@@ -884,6 +923,7 @@ impl Drop for TimelineMetrics {
         let _ = PERSISTENT_BYTES_WRITTEN.remove_label_values(&[tenant_id, timeline_id]);
         let _ = EVICTIONS.remove_label_values(&[tenant_id, timeline_id]);
         let _ = READ_NUM_FS_LAYERS.remove_label_values(&[tenant_id, timeline_id]);
+        let _ = STORAGE_PHYSICAL_SIZE.remove_label_values(&[tenant_id, timeline_id]);
 
         self.evictions_with_low_residence_duration
             .write()
@@ -906,6 +946,9 @@ impl Drop for TimelineMetrics {
         for op in SMGR_QUERY_TIME_OPERATIONS {
             let _ = SMGR_QUERY_TIME.remove_label_values(&[op, tenant_id, timeline_id]);
         }
+        for ty in STORAGE_PHYSICAL_SIZE_FILE_TYPE {
+            let _ = STORAGE_PHYSICAL_SIZE.remove_label_values(&[ty, tenant_id, timeline_id]);
+        }
     }
 }
 

pageserver/src/tenant.rs

@@ -1560,7 +1560,7 @@ impl Tenant {
             // No timeout here, GC & Compaction should be responsive to the
             // `TimelineState::Stopping` change.
             info!("waiting for layer_removal_cs.lock()");
-            let layer_removal_guard = timeline.lcache.delete_guard().await;
+            let layer_removal_guard = timeline.lcache.delete_guard_write().await;
             info!("got layer_removal_cs.lock(), deleting layer files");
 
             // NB: storage_sync upload tasks that reference these layers have been cancelled

pageserver/src/tenant/layer_cache.rs

@@ -1,9 +1,11 @@
 use super::storage_layer::{PersistentLayer, PersistentLayerDesc, PersistentLayerKey, RemoteLayer};
 use super::Timeline;
+use crate::metrics::{STORAGE_PHYSICAL_SIZE, STORAGE_PHYSICAL_SIZE_FILE_TYPE};
 use crate::tenant::layer_map::{self, LayerMap};
 use anyhow::Result;
 use std::sync::{Mutex, Weak};
 use std::{collections::HashMap, sync::Arc};
+use utils::id::{TenantId, TimelineId};
 
 pub struct LayerCache {
     /// Layer removal lock.
@@ -11,7 +13,7 @@ pub struct LayerCache {
     /// This lock is acquired in [`Timeline::gc`], [`Timeline::compact`],
     /// and [`Tenant::delete_timeline`]. This is an `Arc<Mutex>` lock because we need an owned
     /// lock guard in functions that will be spawned to tokio I/O pool (which requires `'static`).
-    pub layers_removal_lock: Arc<tokio::sync::Mutex<()>>,
+    pub layers_removal_lock: Arc<tokio::sync::RwLock<()>>,
 
     /// We need this lock b/c we do not have any way to prevent GC/compaction from removing files in-use.
     /// We need to do reference counting on Arc to prevent this from happening, and we can safely remove this lock.
@@ -21,6 +23,11 @@ pub struct LayerCache {
     #[allow(unused)]
     timeline: Weak<Timeline>,
 
+    pub tenant_id: TenantId,
+    pub timeline_id: TimelineId,
+    pub tenant_id_str: String,
+    pub timeline_id_str: String,
+
     mapping: Mutex<HashMap<PersistentLayerKey, Arc<dyn PersistentLayer>>>,
 }
 
@@ -29,15 +36,22 @@ pub struct LayerInUseWrite(tokio::sync::OwnedRwLockWriteGuard<()>);
 pub struct LayerInUseRead(tokio::sync::OwnedRwLockReadGuard<()>);
 
 #[derive(Clone)]
-pub struct DeleteGuard(Arc<tokio::sync::OwnedMutexGuard<()>>);
+pub struct DeleteGuardRead(Arc<tokio::sync::OwnedRwLockReadGuard<()>>);
+
+#[derive(Clone)]
+pub struct DeleteGuardWrite(Arc<tokio::sync::OwnedRwLockWriteGuard<()>>);
 
 impl LayerCache {
-    pub fn new(timeline: Weak<Timeline>) -> Self {
+    pub fn new(timeline: Weak<Timeline>, tenant_id: TenantId, timeline_id: TimelineId) -> Self {
         Self {
             layers_operation_lock: Arc::new(tokio::sync::RwLock::new(())),
-            layers_removal_lock: Arc::new(tokio::sync::Mutex::new(())),
+            layers_removal_lock: Arc::new(tokio::sync::RwLock::new(())),
             mapping: Mutex::new(HashMap::new()),
-            timeline,
+            timeline: timeline,
+            tenant_id: tenant_id,
+            timeline_id: timeline_id,
+            tenant_id_str: tenant_id.to_string(),
+            timeline_id_str: timeline_id.to_string(),
         }
     }
 
@@ -59,26 +73,36 @@ impl LayerCache {
     }
 
     /// Ensures only one of compaction / gc can happen at a time.
-    pub async fn delete_guard(&self) -> DeleteGuard {
-        DeleteGuard(Arc::new(
-            self.layers_removal_lock.clone().lock_owned().await,
+    pub async fn delete_guard_read(&self) -> DeleteGuardRead {
+        DeleteGuardRead(Arc::new(
+            self.layers_removal_lock.clone().read_owned().await,
+        ))
+    }
+
+    /// Ensures only one of compaction / gc can happen at a time.
+    pub async fn delete_guard_write(&self) -> DeleteGuardWrite {
+        DeleteGuardWrite(Arc::new(
+            self.layers_removal_lock.clone().write_owned().await,
         ))
     }
 
     /// Should only be called when initializing the timeline. Bypass checks and layer operation lock.
     pub fn remove_local_when_init(&self, layer: Arc<dyn PersistentLayer>) {
+        self.metrics_size_sub(&*layer);
         let mut guard = self.mapping.lock().unwrap();
         guard.remove(&layer.layer_desc().key());
     }
 
     /// Should only be called when initializing the timeline. Bypass checks and layer operation lock.
     pub fn populate_remote_when_init(&self, layer: Arc<RemoteLayer>) {
+        self.metrics_size_add(&*layer);
         let mut guard = self.mapping.lock().unwrap();
         guard.insert(layer.layer_desc().key(), layer);
     }
 
     /// Should only be called when initializing the timeline. Bypass checks and layer operation lock.
     pub fn populate_local_when_init(&self, layer: Arc<dyn PersistentLayer>) {
+        self.metrics_size_add(&*layer);
         let mut guard = self.mapping.lock().unwrap();
         guard.insert(layer.layer_desc().key(), layer);
     }
@@ -91,9 +115,8 @@ impl LayerCache {
     ) -> Result<()> {
         let mut guard = self.mapping.lock().unwrap();
 
-        use super::layer_map::LayerKey;
-        let key = LayerKey::from(&*expected);
-        let other = LayerKey::from(&*new);
+        let key: PersistentLayerKey = expected.layer_desc().key();
+        let other = new.layer_desc().key();
 
         let expected_l0 = LayerMap::is_l0(expected.layer_desc());
         let new_l0 = LayerMap::is_l0(new.layer_desc());
@@ -130,6 +153,7 @@ impl LayerCache {
 
     /// Called within write path. When compaction and image layer creation we will create new layers.
     pub fn create_new_layer(&self, layer: Arc<dyn PersistentLayer>) {
+        self.metrics_size_add(&*layer);
         let mut guard = self.mapping.lock().unwrap();
         guard.insert(layer.layer_desc().key(), layer);
     }
@@ -137,7 +161,38 @@ impl LayerCache {
     /// Called within write path. When GC and compaction we will remove layers and delete them on disk.
     /// Will move logic to delete files here later.
     pub fn delete_layer(&self, layer: Arc<dyn PersistentLayer>) {
+        self.metrics_size_sub(&*layer);
         let mut guard = self.mapping.lock().unwrap();
         guard.remove(&layer.layer_desc().key());
     }
+
+    fn metrics_size_add(&self, layer: &dyn PersistentLayer) {
+        STORAGE_PHYSICAL_SIZE
+            .with_label_values(&[
+                Self::get_layer_type(layer),
+                &self.tenant_id_str,
+                &self.timeline_id_str,
+            ])
+            .add(layer.file_size() as i64);
+    }
+
+    fn metrics_size_sub(&self, layer: &dyn PersistentLayer) {
+        STORAGE_PHYSICAL_SIZE
+            .with_label_values(&[
+                Self::get_layer_type(layer),
+                &self.tenant_id_str,
+                &self.timeline_id_str,
+            ])
+            .sub(layer.file_size() as i64);
+    }
+
+    fn get_layer_type(layer: &dyn PersistentLayer) -> &'static str {
+        if layer.layer_desc().is_delta() {
+            &STORAGE_PHYSICAL_SIZE_FILE_TYPE[1]
+        } else if layer.layer_desc().is_incremental() {
+            &STORAGE_PHYSICAL_SIZE_FILE_TYPE[2]
+        } else {
+            &STORAGE_PHYSICAL_SIZE_FILE_TYPE[0]
+        }
+    }
 }

pageserver/src/tenant/layer_map.rs

@@ -93,6 +93,58 @@ pub struct LayerMap {
     /// L0 layers have key range Key::MIN..Key::MAX, and locating them using R-Tree search is very inefficient.
     /// So L0 layers are held in l0_delta_layers vector, in addition to the R-tree.
     l0_delta_layers: Vec<Arc<PersistentLayerDesc>>,
+
+    /// All sorted runs. For tiered compaction.
+    pub sorted_runs: SortedRuns,
+}
+
+#[derive(Default)]
+pub struct SortedRuns {
+    pub runs: Vec<(usize, Vec<Arc<PersistentLayerDesc>>)>,
+    next_tier_id: usize,
+}
+
+impl SortedRuns {
+    /// Create a new sorted run and insert it at the top of the LSM tree.
+    pub fn create_new_run(&mut self, layers: Vec<Arc<PersistentLayerDesc>>) -> usize {
+        let tier_id = self.next_tier_id();
+        self.runs.insert(0, (tier_id, layers));
+        tier_id
+    }
+
+    /// Create a new sorted run and insert it at the bottom of the LSM tree.
+    pub fn create_new_bottom_run(&mut self, layers: Vec<Arc<PersistentLayerDesc>>) -> usize {
+        let tier_id = self.next_tier_id();
+        self.runs.push((tier_id, layers));
+        tier_id
+    }
+
+    pub fn compute_tier_sizes(&self) -> Vec<(usize, u64)> {
+        self.runs
+            .iter()
+            .map(|(tier_id, layers)| (*tier_id, layers.iter().map(|layer| layer.file_size()).sum()))
+            .collect::<Vec<_>>()
+    }
+
+    /// Remove a sorted run from the LSM tree.
+    pub fn remove_run(&mut self, tier_id: usize) {
+        self.runs.retain(|(id, _)| *id != tier_id);
+    }
+
+    /// Remove layers and the corresponding sorted runs.
+    pub fn insert_run_at(&mut self, idx: usize, layers: Vec<Arc<PersistentLayerDesc>>) {
+        unimplemented!()
+    }
+
+    pub fn num_of_tiers(&self) -> usize {
+        self.runs.len()
+    }
+
+    pub fn next_tier_id(&mut self) -> usize {
+        let ret = self.next_tier_id;
+        self.next_tier_id += 1;
+        ret
+    }
 }
 
 /// The primary update API for the layer map.
@@ -114,15 +166,28 @@ impl BatchedUpdates<'_> {
     ///
     // TODO remove the `layer` argument when `mapping` is refactored out of `LayerMap`
     pub fn insert_historic(&mut self, layer_desc: PersistentLayerDesc) {
+        self.insert_historic_new(layer_desc) // insert into layer map without populating tiering structure
+    }
+
+    pub fn insert_historic_new(&mut self, layer_desc: PersistentLayerDesc) {
         self.layer_map.insert_historic_noflush(layer_desc)
     }
 
+    /// Get a reference to the current sorted runs.
+    pub fn sorted_runs(&mut self) -> &mut SortedRuns {
+        &mut self.layer_map.sorted_runs
+    }
+
     ///
     /// Remove an on-disk layer from the map.
     ///
     /// This should be called when the corresponding file on disk has been deleted.
     ///
     pub fn remove_historic(&mut self, layer_desc: PersistentLayerDesc) {
+        self.remove_historic_new(layer_desc) // remove from layer map without populating tiering structure
+    }
+
+    pub fn remove_historic_new(&mut self, layer_desc: PersistentLayerDesc) {
         self.layer_map.remove_historic_noflush(layer_desc)
     }
 
@@ -184,14 +249,28 @@ impl LayerMap {
     /// 'open' and 'frozen' layers!
     ///
     pub fn search(&self, key: Key, end_lsn: Lsn) -> Option<SearchResult> {
+        self.search_incremental(key, end_lsn, false)
+    }
+
+    pub fn search_incremental(
+        &self,
+        key: Key,
+        end_lsn: Lsn,
+        exclude_image: bool,
+    ) -> Option<SearchResult> {
         let version = self.historic.get().unwrap().get_version(end_lsn.0 - 1)?;
         let latest_delta = version.delta_coverage.query(key.to_i128());
-        let latest_image = version.image_coverage.query(key.to_i128());
+        let latest_image = if exclude_image {
+            let version = self.historic.get().unwrap().get_version(end_lsn.0 - 2)?;
+            version.image_coverage.query(key.to_i128())
+        } else {
+            version.image_coverage.query(key.to_i128())
+        };
 
         match (latest_delta, latest_image) {
             (None, None) => None,
             (None, Some(image)) => {
-                let lsn_floor = image.get_lsn_range().start;
+                let lsn_floor = image.get_lsn_range().end;
                 Some(SearchResult {
                     layer: image,
                     lsn_floor,
@@ -211,7 +290,7 @@ impl LayerMap {
                 if image_is_newer || image_exact_match {
                     Some(SearchResult {
                         layer: image,
-                        lsn_floor: img_lsn,
+                        lsn_floor: img_lsn + 1,
                     })
                 } else {
                     let lsn_floor =
@@ -640,10 +719,19 @@ impl LayerMap {
             frozen_layer.dump(verbose, ctx)?;
         }
 
-        println!("historic_layers:");
-        for layer in self.iter_historic_layers() {
+        println!("l0_deltas:");
+        for layer in &self.l0_delta_layers {
             layer.dump(verbose, ctx)?;
         }
+
+        println!("sorted_runs:");
+        for (lvl, (tier_id, layer)) in self.sorted_runs.runs.iter().enumerate() {
+            println!("tier {}", tier_id);
+            for layer in layer {
+                layer.dump(verbose, ctx)?;
+            }
+        }
+
         println!("End dump LayerMap");
         Ok(())
     }
@@ -691,6 +779,7 @@ mod tests {
         use super::*;
 
         #[test]
+        #[ignore]
         fn for_full_range_delta() {
             // l0_delta_layers are used by compaction, and should observe all buffered updates
             l0_delta_layers_updated_scenario(
@@ -700,6 +789,7 @@ mod tests {
         }
 
         #[test]
+        #[ignore]
         fn for_non_full_range_delta() {
             // has minimal uncovered areas compared to l0_delta_layers_updated_on_insert_replace_remove_for_full_range_delta
             l0_delta_layers_updated_scenario(
@@ -710,6 +800,7 @@ mod tests {
         }
 
         #[test]
+        #[ignore]
         fn for_image() {
             l0_delta_layers_updated_scenario(
                  "000000000000000000000000000000000000-000000000000000000000000000000010000__0000000053424D69",

pageserver/src/tenant/layer_map/historic_layer_coverage.rs

@@ -43,18 +43,6 @@ impl Ord for LayerKey {
     }
 }
 
-impl<'a, L: crate::tenant::storage_layer::Layer + ?Sized> From<&'a L> for LayerKey {
-    fn from(layer: &'a L) -> Self {
-        let kr = layer.get_key_range();
-        let lr = layer.get_lsn_range();
-        LayerKey {
-            key: kr.start.to_i128()..kr.end.to_i128(),
-            lsn: lr.start.0..lr.end.0,
-            is_image: !layer.is_incremental(),
-        }
-    }
-}
-
 impl From<&PersistentLayerDesc> for LayerKey {
     fn from(layer: &PersistentLayerDesc) -> Self {
         let kr = layer.get_key_range();
@@ -62,7 +50,7 @@ impl From<&PersistentLayerDesc> for LayerKey {
         LayerKey {
             key: kr.start.to_i128()..kr.end.to_i128(),
             lsn: lr.start.0..lr.end.0,
-            is_image: !layer.is_incremental(),
+            is_image: !layer.is_delta,
         }
     }
 }

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -222,13 +222,14 @@ impl Layer for DeltaLayer {
     /// debugging function to print out the contents of the layer
     fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
         println!(
-            "----- delta layer for ten {} tli {} keys {}-{} lsn {}-{} ----",
+            "----- delta layer for ten {} tli {} keys {}-{} lsn {}-{} size {} ----",
             self.desc.tenant_id,
             self.desc.timeline_id,
             self.desc.key_range.start,
             self.desc.key_range.end,
             self.desc.lsn_range.start,
-            self.desc.lsn_range.end
+            self.desc.lsn_range.end,
+            self.desc.file_size
         );
 
         if !verbose {

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -153,12 +153,13 @@ impl Layer for ImageLayer {
     /// debugging function to print out the contents of the layer
     fn dump(&self, verbose: bool, ctx: &RequestContext) -> Result<()> {
         println!(
-            "----- image layer for ten {} tli {} key {}-{} at {} ----",
+            "----- image layer for ten {} tli {} key {}-{} at {} size {} ----",
             self.desc.tenant_id,
             self.desc.timeline_id,
             self.desc.key_range.start,
             self.desc.key_range.end,
-            self.lsn
+            self.lsn,
+            self.desc.file_size
         );
 
         if !verbose {
@@ -212,7 +213,11 @@ impl Layer for ImageLayer {
             reconstruct_state.img = Some((self.lsn, value));
             Ok(ValueReconstructResult::Complete)
         } else {
-            Ok(ValueReconstructResult::Missing)
+            if self.desc.is_incremental {
+                Ok(ValueReconstructResult::Continue)
+            } else {
+                Ok(ValueReconstructResult::Missing)
+            }
         }
     }
 
@@ -404,7 +409,7 @@ impl ImageLayer {
                 timeline_id,
                 filename.key_range.clone(),
                 filename.lsn,
-                false,
+                true,
                 file_size,
             ), // Now we assume image layer ALWAYS covers the full range. This may change in the future.
             lsn: filename.lsn,
@@ -436,7 +441,7 @@ impl ImageLayer {
                 summary.timeline_id,
                 summary.key_range,
                 summary.lsn,
-                false,
+                true,
                 metadata.len(),
             ), // Now we assume image layer ALWAYS covers the full range. This may change in the future.
             lsn: summary.lsn,
@@ -481,12 +486,14 @@ struct ImageLayerWriterInner {
     path: PathBuf,
     timeline_id: TimelineId,
     tenant_id: TenantId,
-    key_range: Range<Key>,
     lsn: Lsn,
     is_incremental: bool,
 
     blob_writer: WriteBlobWriter<VirtualFile>,
     tree: DiskBtreeBuilder<BlockBuf, KEY_SIZE>,
+
+    start_key: Key,
+    last_key: Option<Key>,
 }
 
 impl ImageLayerWriterInner {
@@ -497,8 +504,8 @@ impl ImageLayerWriterInner {
         conf: &'static PageServerConf,
         timeline_id: TimelineId,
         tenant_id: TenantId,
-        key_range: &Range<Key>,
         lsn: Lsn,
+        start_key: Key,
         is_incremental: bool,
     ) -> anyhow::Result<Self> {
         // Create the file initially with a temporary filename.
@@ -508,7 +515,7 @@ impl ImageLayerWriterInner {
             timeline_id,
             tenant_id,
             &ImageFileName {
-                key_range: key_range.clone(),
+                key_range: start_key..start_key, // TODO(chi): use number instead of dummy range
                 lsn,
             },
         );
@@ -530,11 +537,12 @@ impl ImageLayerWriterInner {
             path,
             timeline_id,
             tenant_id,
-            key_range: key_range.clone(),
             lsn,
             tree: tree_builder,
             blob_writer,
             is_incremental,
+            start_key,
+            last_key: None,
         };
 
         Ok(writer)
@@ -546,7 +554,14 @@ impl ImageLayerWriterInner {
     /// The page versions must be appended in blknum order.
     ///
     fn put_image(&mut self, key: Key, img: &[u8]) -> anyhow::Result<()> {
-        ensure!(self.key_range.contains(&key));
+        if cfg!(debug_assertions) {
+            ensure!(key >= self.start_key);
+            if let Some(last_key) = self.last_key.as_ref() {
+                ensure!(last_key < &key);
+            }
+            self.last_key = Some(key.clone());
+        }
+
         let off = self.blob_writer.write_blob(img)?;
 
         let mut keybuf: [u8; KEY_SIZE] = [0u8; KEY_SIZE];
@@ -559,7 +574,7 @@ impl ImageLayerWriterInner {
     ///
     /// Finish writing the image layer.
     ///
-    fn finish(self) -> anyhow::Result<ImageLayer> {
+    fn finish(self, end_key: Key) -> anyhow::Result<ImageLayer> {
         let index_start_blk =
             ((self.blob_writer.size() + PAGE_SZ as u64 - 1) / PAGE_SZ as u64) as u32;
 
@@ -572,13 +587,15 @@ impl ImageLayerWriterInner {
             file.write_all(buf.as_ref())?;
         }
 
+        let key_range = self.start_key.clone()..end_key;
+
         // Fill in the summary on blk 0
         let summary = Summary {
             magic: IMAGE_FILE_MAGIC,
             format_version: STORAGE_FORMAT_VERSION,
             tenant_id: self.tenant_id,
             timeline_id: self.timeline_id,
-            key_range: self.key_range.clone(),
+            key_range: key_range.clone(),
             lsn: self.lsn,
             index_start_blk,
             index_root_blk,
@@ -593,7 +610,7 @@ impl ImageLayerWriterInner {
         let desc = PersistentLayerDesc::new_img(
             self.tenant_id,
             self.timeline_id,
-            self.key_range.clone(),
+            key_range.clone(),
             self.lsn,
             self.is_incremental, // for now, image layer ALWAYS covers the full range
             metadata.len(),
@@ -627,7 +644,7 @@ impl ImageLayerWriterInner {
             self.timeline_id,
             self.tenant_id,
             &ImageFileName {
-                key_range: self.key_range.clone(),
+                key_range,
                 lsn: self.lsn,
             },
         );
@@ -637,6 +654,10 @@ impl ImageLayerWriterInner {
 
         Ok(layer)
     }
+
+    fn size(&self) -> u64 {
+        self.blob_writer.size() + self.tree.borrow_writer().size()
+    }
 }
 
 /// A builder object for constructing a new image layer.
@@ -673,7 +694,7 @@ impl ImageLayerWriter {
         conf: &'static PageServerConf,
         timeline_id: TimelineId,
         tenant_id: TenantId,
-        key_range: &Range<Key>,
+        start_key: Key,
         lsn: Lsn,
         is_incremental: bool,
     ) -> anyhow::Result<ImageLayerWriter> {
@@ -682,8 +703,8 @@ impl ImageLayerWriter {
                 conf,
                 timeline_id,
                 tenant_id,
-                key_range,
                 lsn,
+                start_key,
                 is_incremental,
             )?),
         })
@@ -701,8 +722,12 @@ impl ImageLayerWriter {
     ///
     /// Finish writing the image layer.
     ///
-    pub fn finish(mut self) -> anyhow::Result<ImageLayer> {
-        self.inner.take().unwrap().finish()
+    pub fn finish(mut self, end_key: Key) -> anyhow::Result<ImageLayer> {
+        self.inner.take().unwrap().finish(end_key)
+    }
+
+    pub fn size(&self) -> u64 {
+        self.inner.as_ref().unwrap().size()
     }
 }
 

pageserver/src/tenant/storage_layer/inmemory_layer.rs

@@ -11,6 +11,7 @@ use crate::tenant::blob_io::{BlobCursor, BlobWriter};
 use crate::tenant::block_io::BlockReader;
 use crate::tenant::ephemeral_file::EphemeralFile;
 use crate::tenant::storage_layer::{ValueReconstructResult, ValueReconstructState};
+use crate::tenant::timeline::ENABLE_TIERED_COMPACTION;
 use crate::walrecord;
 use anyhow::{ensure, Result};
 use pageserver_api::models::InMemoryLayerInfo;
@@ -149,8 +150,8 @@ impl Layer for InMemoryLayer {
             .unwrap_or_default();
 
         println!(
-            "----- in-memory layer for tli {} LSNs {}-{} ----",
-            self.timeline_id, self.start_lsn, end_str,
+            "----- in-memory layer LSNs {}-{} ----",
+            self.start_lsn, end_str,
         );
 
         if !verbose {
@@ -341,11 +342,18 @@ impl InMemoryLayer {
         // rare though, so we just accept the potential latency hit for now.
         let inner = self.inner.read().unwrap();
 
+        let mut keys: Vec<(&Key, &VecMap<Lsn, u64>)> = inner.index.iter().collect();
+        keys.sort_by_key(|k| k.0);
+
         let mut delta_layer_writer = DeltaLayerWriter::new(
             self.conf,
             self.timeline_id,
             self.tenant_id,
-            Key::MIN,
+            if ENABLE_TIERED_COMPACTION {
+                keys.first().unwrap().0.clone()
+            } else {
+                Key::MIN
+            },
             self.start_lsn..inner.end_lsn.unwrap(),
         )?;
 
@@ -353,9 +361,6 @@ impl InMemoryLayer {
 
         let mut cursor = inner.file.block_cursor();
 
-        let mut keys: Vec<(&Key, &VecMap<Lsn, u64>)> = inner.index.iter().collect();
-        keys.sort_by_key(|k| k.0);
-
         for (key, vec_map) in keys.iter() {
             let key = **key;
             // Write all page versions
@@ -366,7 +371,11 @@ impl InMemoryLayer {
             }
         }
 
-        let delta_layer = delta_layer_writer.finish(Key::MAX)?;
+        let delta_layer = delta_layer_writer.finish(if ENABLE_TIERED_COMPACTION {
+            keys.last().unwrap().0.next()
+        } else {
+            Key::MAX
+        })?;
         Ok(delta_layer)
     }
 }

pageserver/src/tenant/storage_layer/layer_desc.rs

@@ -173,13 +173,14 @@ impl PersistentLayerDesc {
 
     pub fn dump(&self, _verbose: bool, _ctx: &RequestContext) -> Result<()> {
         println!(
-            "----- layer for ten {} tli {} keys {}-{} lsn {}-{} ----",
-            self.tenant_id,
-            self.timeline_id,
+            "----- layer for keys {}-{} lsn {}-{} size {} is_delta {} is_incremental {} ----",
             self.key_range.start,
             self.key_range.end,
             self.lsn_range.start,
-            self.lsn_range.end
+            self.lsn_range.end,
+            self.file_size,
+            self.is_delta,
+            self.is_incremental
         );
 
         Ok(())

pageserver/src/tenant/tasks.rs

@@ -14,35 +14,43 @@ use tokio_util::sync::CancellationToken;
 use tracing::*;
 use utils::completion;
 
+use super::timeline::ENABLE_TIERED_COMPACTION;
+
 /// Start per tenant background loops: compaction and gc.
 pub fn start_background_loops(
     tenant: &Arc<Tenant>,
     background_jobs_can_start: Option<&completion::Barrier>,
 ) {
     let tenant_id = tenant.tenant_id;
-    task_mgr::spawn(
-        BACKGROUND_RUNTIME.handle(),
-        TaskKind::Compaction,
-        Some(tenant_id),
-        None,
-        &format!("compactor for tenant {tenant_id}"),
-        false,
-        {
-            let tenant = Arc::clone(tenant);
-            let background_jobs_can_start = background_jobs_can_start.cloned();
-            async move {
-                let cancel = task_mgr::shutdown_token();
-                tokio::select! {
-                    _ = cancel.cancelled() => { return Ok(()) },
-                    _ = completion::Barrier::maybe_wait(background_jobs_can_start) => {}
-                };
-                compaction_loop(tenant, cancel)
-                    .instrument(info_span!("compaction_loop", tenant_id = %tenant_id))
-                    .await;
-                Ok(())
-            }
-        },
-    );
+    // start two compaction threads
+    let range = if ENABLE_TIERED_COMPACTION { 0..4 } else { 0..1 };
+    for cpt_id in range {
+        task_mgr::spawn(
+            BACKGROUND_RUNTIME.handle(),
+            TaskKind::Compaction,
+            Some(tenant_id),
+            None,
+            &format!("compactor for tenant {tenant_id}"),
+            false,
+            {
+                let tenant = Arc::clone(tenant);
+                let background_jobs_can_start = background_jobs_can_start.cloned();
+                async move {
+                    let cancel = task_mgr::shutdown_token();
+                    tokio::select! {
+                        _ = cancel.cancelled() => { return Ok(()) },
+                        _ = completion::Barrier::maybe_wait(background_jobs_can_start) => {}
+                    };
+                    compaction_loop(tenant, cancel)
+                        .instrument(
+                            info_span!("compaction_loop", tenant_id = %tenant_id, cpt_id = %cpt_id),
+                        )
+                        .await;
+                    Ok(())
+                }
+            },
+        );
+    }
     task_mgr::spawn(
         BACKGROUND_RUNTIME.handle(),
         TaskKind::GarbageCollector,