Add ps compaction_threshold config

Signed-off-by: Dhammika Pathirana <dhammika@gmail.com>

Add ps compaction_threadhold knob for (#707) (#1484)

pageserver/src/config.rs

@@ -36,8 +36,8 @@ pub mod defaults {
     // Target file size, when creating image and delta layers.
     // This parameter determines L1 layer file size.
     pub const DEFAULT_COMPACTION_TARGET_SIZE: u64 = 128 * 1024 * 1024;
-
     pub const DEFAULT_COMPACTION_PERIOD: &str = "1 s";
+    pub const DEFAULT_COMPACTION_THRESHOLD: usize = 10;
 
     pub const DEFAULT_GC_HORIZON: u64 = 64 * 1024 * 1024;
     pub const DEFAULT_GC_PERIOD: &str = "100 s";
@@ -65,6 +65,7 @@ pub mod defaults {
 #checkpoint_distance = {DEFAULT_CHECKPOINT_DISTANCE} # in bytes
 #compaction_target_size = {DEFAULT_COMPACTION_TARGET_SIZE} # in bytes
 #compaction_period = '{DEFAULT_COMPACTION_PERIOD}'
+#compaction_threshold = '{DEFAULT_COMPACTION_THRESHOLD}'
 
 #gc_period = '{DEFAULT_GC_PERIOD}'
 #gc_horizon = {DEFAULT_GC_HORIZON}
@@ -107,6 +108,9 @@ pub struct PageServerConf {
     // How often to check if there's compaction work to be done.
     pub compaction_period: Duration,
 
+    // Level0 delta layer threshold for compaction.
+    pub compaction_threshold: usize,
+
     pub gc_horizon: u64,
     pub gc_period: Duration,
 
@@ -162,6 +166,7 @@ struct PageServerConfigBuilder {
 
     compaction_target_size: BuilderValue<u64>,
     compaction_period: BuilderValue<Duration>,
+    compaction_threshold: BuilderValue<usize>,
 
     gc_horizon: BuilderValue<u64>,
     gc_period: BuilderValue<Duration>,
@@ -198,6 +203,7 @@ impl Default for PageServerConfigBuilder {
             compaction_target_size: Set(DEFAULT_COMPACTION_TARGET_SIZE),
             compaction_period: Set(humantime::parse_duration(DEFAULT_COMPACTION_PERIOD)
                 .expect("cannot parse default compaction period")),
+            compaction_threshold: Set(DEFAULT_COMPACTION_THRESHOLD),
             gc_horizon: Set(DEFAULT_GC_HORIZON),
             gc_period: Set(humantime::parse_duration(DEFAULT_GC_PERIOD)
                 .expect("cannot parse default gc period")),
@@ -241,6 +247,10 @@ impl PageServerConfigBuilder {
         self.compaction_period = BuilderValue::Set(compaction_period)
     }
 
+    pub fn compaction_threshold(&mut self, compaction_threshold: usize) {
+        self.compaction_threshold = BuilderValue::Set(compaction_threshold)
+    }
+
     pub fn gc_horizon(&mut self, gc_horizon: u64) {
         self.gc_horizon = BuilderValue::Set(gc_horizon)
     }
@@ -313,6 +323,9 @@ impl PageServerConfigBuilder {
             compaction_period: self
                 .compaction_period
                 .ok_or(anyhow::anyhow!("missing compaction_period"))?,
+            compaction_threshold: self
+                .compaction_threshold
+                .ok_or(anyhow::anyhow!("missing compaction_threshold"))?,
             gc_horizon: self
                 .gc_horizon
                 .ok_or(anyhow::anyhow!("missing gc_horizon"))?,
@@ -453,6 +466,9 @@ impl PageServerConf {
                     builder.compaction_target_size(parse_toml_u64(key, item)?)
                 }
                 "compaction_period" => builder.compaction_period(parse_toml_duration(key, item)?),
+                "compaction_threshold" => {
+                    builder.compaction_threshold(parse_toml_u64(key, item)? as usize)
+                }
                 "gc_horizon" => builder.gc_horizon(parse_toml_u64(key, item)?),
                 "gc_period" => builder.gc_period(parse_toml_duration(key, item)?),
                 "wait_lsn_timeout" => builder.wait_lsn_timeout(parse_toml_duration(key, item)?),
@@ -590,6 +606,7 @@ impl PageServerConf {
             checkpoint_distance: defaults::DEFAULT_CHECKPOINT_DISTANCE,
             compaction_target_size: 4 * 1024 * 1024,
             compaction_period: Duration::from_secs(10),
+            compaction_threshold: defaults::DEFAULT_COMPACTION_THRESHOLD,
             gc_horizon: defaults::DEFAULT_GC_HORIZON,
             gc_period: Duration::from_secs(10),
             wait_lsn_timeout: Duration::from_secs(60),
@@ -662,6 +679,7 @@ checkpoint_distance = 111 # in bytes
 
 compaction_target_size = 111 # in bytes
 compaction_period = '111 s'
+compaction_threshold = 2
 
 gc_period = '222 s'
 gc_horizon = 222
@@ -700,6 +718,7 @@ id = 10
                 checkpoint_distance: defaults::DEFAULT_CHECKPOINT_DISTANCE,
                 compaction_target_size: defaults::DEFAULT_COMPACTION_TARGET_SIZE,
                 compaction_period: humantime::parse_duration(defaults::DEFAULT_COMPACTION_PERIOD)?,
+                compaction_threshold: defaults::DEFAULT_COMPACTION_THRESHOLD,
                 gc_horizon: defaults::DEFAULT_GC_HORIZON,
                 gc_period: humantime::parse_duration(defaults::DEFAULT_GC_PERIOD)?,
                 wait_lsn_timeout: humantime::parse_duration(defaults::DEFAULT_WAIT_LSN_TIMEOUT)?,
@@ -745,6 +764,7 @@ id = 10
                 checkpoint_distance: 111,
                 compaction_target_size: 111,
                 compaction_period: Duration::from_secs(111),
+                compaction_threshold: 2,
                 gc_horizon: 222,
                 gc_period: Duration::from_secs(222),
                 wait_lsn_timeout: Duration::from_secs(111),

pageserver/src/layered_repository.rs

@@ -1680,13 +1680,11 @@ impl LayeredTimeline {
     fn compact_level0(&self, target_file_size: u64) -> Result<()> {
         let layers = self.layers.lock().unwrap();
 
-        // We compact or "shuffle" the level-0 delta layers when 10 have
-        // accumulated.
-        static COMPACT_THRESHOLD: usize = 10;
-
         let level0_deltas = layers.get_level0_deltas()?;
 
-        if level0_deltas.len() < COMPACT_THRESHOLD {
+        // We compact or "shuffle" the level-0 delta layers when they've
+        // accumulated over the compaction threshold.
+        if level0_deltas.len() < self.conf.compaction_threshold {
             return Ok(());
         }
         drop(layers);