Merge pull request tursodatabase#1782 from tursodatabase/tiered-compa…

…ction

tiered compaction

libsql-server/src/wal_toolkit.rs

@@ -12,16 +12,11 @@ use libsql_wal::storage::backend::s3::S3Backend;
 use libsql_wal::storage::backend::Backend;
 use libsql_wal::storage::compaction::strategy::identity::IdentityStrategy;
 use libsql_wal::storage::compaction::strategy::log_strategy::LogReductionStrategy;
-use libsql_wal::storage::compaction::strategy::PartitionStrategy;
+use libsql_wal::storage::compaction::strategy::tiered::LevelsStrategy;
+use libsql_wal::storage::compaction::strategy::CompactionStrategy;
 use libsql_wal::storage::compaction::Compactor;
 use rusqlite::OpenFlags;
 
-#[derive(Clone, Debug, clap::ValueEnum, Copy)]
-pub enum CompactStrategy {
-    Logarithmic,
-    CompactAll,
-}
-
 #[derive(Debug, clap::Subcommand)]
 pub enum WalToolkitCommand {
     Monitor(MonitorCommand),
@@ -119,6 +114,13 @@ impl SyncCommand {
     }
 }
 
+#[derive(Clone, Debug, clap::ValueEnum, Copy)]
+pub enum CompactStrategy {
+    Logarithmic,
+    CompactAll,
+    Tiered,
+}
+
 #[derive(Debug, clap::Args)]
 /// Compact segments into bigger segments
 pub struct CompactCommand {
@@ -168,10 +170,12 @@ impl CompactCommand {
         namespace: &NamespaceName,
     ) -> anyhow::Result<()> {
         let analysis = compactor.analyze(&namespace)?;
-        let strat: Box<dyn PartitionStrategy> = match self.strategy {
+        let strat: Box<dyn CompactionStrategy> = match self.strategy {
             CompactStrategy::Logarithmic => Box::new(LogReductionStrategy),
             CompactStrategy::CompactAll => Box::new(IdentityStrategy),
+            CompactStrategy::Tiered => Box::new(LevelsStrategy::new(self.threshold)),
         };
+
         let set = analysis.shortest_restore_path();
         if set.len() <= self.threshold {
             println!(

libsql-wal/src/shared_wal.rs

@@ -51,7 +51,6 @@ pub struct SharedWal<IO: Io> {
     pub(crate) db_file: IO::File,
     pub(crate) namespace: NamespaceName,
     pub(crate) registry: Arc<dyn SwapLog<IO>>,
-    #[allow(dead_code)] // used by replication
     pub(crate) checkpointed_frame_no: AtomicU64,
     /// max frame_no acknowledged by the durable storage
     pub(crate) durable_frame_no: Arc<Mutex<u64>>,

libsql-wal/src/storage/compaction/mod.rs

@@ -591,13 +591,23 @@ impl AnalyzedSegments {
 
 /// A set of segments, with the guarantee that segments are non-overlapping and increasing in
 /// frameno
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct SegmentSet {
     namespace: NamespaceName,
     segments: Vec<SegmentKey>,
 }
 
 impl SegmentSet {
+    /// return segments end - start
+    pub fn span(&self) -> u64 {
+        if self.is_empty() {
+            0
+        } else {
+            self.segments.last().unwrap().end_frame_no
+                - self.segments.first().unwrap().start_frame_no
+        }
+    }
+
     pub fn range(&self) -> Option<(u64, u64)> {
         self.segments
             .first()

libsql-wal/src/storage/compaction/strategy/identity.rs

@@ -1,11 +1,11 @@
 use crate::storage::compaction::SegmentSet;
 
-use super::PartitionStrategy;
+use super::CompactionStrategy;
 
 /// partition strategy that doesn't split the passed set
 pub struct IdentityStrategy;
 
-impl PartitionStrategy for IdentityStrategy {
+impl CompactionStrategy for IdentityStrategy {
     fn partition(&self, segments: &SegmentSet) -> Vec<SegmentSet> {
         let mut out = Vec::with_capacity(1);
         out.push(segments.clone());

libsql-wal/src/storage/compaction/strategy/log_strategy.rs

@@ -2,12 +2,12 @@ use std::ops::Deref as _;
 
 use crate::storage::compaction::SegmentSet;
 
-use super::PartitionStrategy;
+use super::CompactionStrategy;
 
 /// partition the SegmentSet in logarithmically reducing sets
 pub struct LogReductionStrategy;
 
-impl PartitionStrategy for LogReductionStrategy {
+impl CompactionStrategy for LogReductionStrategy {
     fn partition(&self, segments: &SegmentSet) -> Vec<SegmentSet> {
         let mut segs = segments.deref();
         let mut out = Vec::new();

libsql-wal/src/storage/compaction/strategy/mod.rs

@@ -2,7 +2,8 @@ use super::SegmentSet;
 
 pub mod identity;
 pub mod log_strategy;
+pub mod tiered;
 
-pub trait PartitionStrategy {
+pub trait CompactionStrategy {
     fn partition(&self, segments: &SegmentSet) -> Vec<SegmentSet>;
 }

libsql-wal/src/storage/compaction/strategy/snapshots/libsql_wal__storage__compaction__strategy__tier_reduction__test__partition_tiered-2.snap

@@ -0,0 +1,34 @@
+---
+source: libsql-wal/src/storage/compaction/strategy/tier_reduction.rs
+expression: partition.first().unwrap()
+---
+SegmentSet {
+    namespace: test,
+    segments: [
+        SegmentKey {
+            start_frame_no: 101,
+            end_frame_no: 105,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 106,
+            end_frame_no: 110,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 111,
+            end_frame_no: 115,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 116,
+            end_frame_no: 120,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 121,
+            end_frame_no: 122,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+    ],
+}

libsql-wal/src/storage/compaction/strategy/snapshots/libsql_wal__storage__compaction__strategy__tier_reduction__test__partition_tiered.snap

@@ -0,0 +1,39 @@
+---
+source: libsql-wal/src/storage/compaction/strategy/tier_reduction.rs
+expression: partition.first().unwrap()
+---
+SegmentSet {
+    namespace: test,
+    segments: [
+        SegmentKey {
+            start_frame_no: 1,
+            end_frame_no: 20,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 21,
+            end_frame_no: 27,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 28,
+            end_frame_no: 41,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 42,
+            end_frame_no: 70,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 71,
+            end_frame_no: 81,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 82,
+            end_frame_no: 100,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+    ],
+}

libsql-wal/src/storage/compaction/strategy/snapshots/libsql_wal__storage__compaction__strategy__tiered__test__partition_tiered-2.snap

@@ -0,0 +1,34 @@
+---
+source: libsql-wal/src/storage/compaction/strategy/tiered.rs
+expression: partition.first().unwrap()
+---
+SegmentSet {
+    namespace: test,
+    segments: [
+        SegmentKey {
+            start_frame_no: 101,
+            end_frame_no: 105,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 106,
+            end_frame_no: 110,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 111,
+            end_frame_no: 115,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 116,
+            end_frame_no: 120,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 121,
+            end_frame_no: 122,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+    ],
+}

libsql-wal/src/storage/compaction/strategy/snapshots/libsql_wal__storage__compaction__strategy__tiered__test__partition_tiered.snap

@@ -0,0 +1,39 @@
+---
+source: libsql-wal/src/storage/compaction/strategy/tiered.rs
+expression: partition.first().unwrap()
+---
+SegmentSet {
+    namespace: test,
+    segments: [
+        SegmentKey {
+            start_frame_no: 1,
+            end_frame_no: 20,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 21,
+            end_frame_no: 27,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 28,
+            end_frame_no: 41,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 42,
+            end_frame_no: 70,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 71,
+            end_frame_no: 81,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+        SegmentKey {
+            start_frame_no: 82,
+            end_frame_no: 100,
+            timestamp: 1970-01-01T00:00:00Z,
+        },
+    ],
+}