fix(storage): do not compact just compacted result

proto/hummock.proto

@@ -28,6 +28,7 @@ message SstableInfo {
   uint64 uncompressed_file_size = 11;
   uint64 range_tombstone_count = 12;
   BloomFilterType bloom_filter_kind = 13;
+  uint64 create_time = 14;
 }
 
 enum LevelType {
@@ -844,6 +845,7 @@ message CompactionConfig {
   uint32 level0_overlapping_sub_level_compact_level_count = 18;
   uint32 tombstone_reclaim_ratio = 19;
   bool enable_emergency_picker = 20;
+  uint64 min_keep_alive_time_secs = 21;
 }
 
 message TableStats {

src/common/src/config.rs

@@ -1667,6 +1667,10 @@ pub mod default {
         pub fn enable_emergency_picker() -> bool {
             DEFAULT_EMERGENCY_PICKER
         }
+
+        pub fn min_keep_alive_time_secs() -> u64 {
+            1200 // 20min
+        }
     }
 
     pub mod object_store_config {

src/meta/src/hummock/compaction/compaction_config.rs

@@ -66,6 +66,7 @@ impl CompactionConfigBuilder {
                     compaction_config::level0_overlapping_sub_level_compact_level_count(),
                 tombstone_reclaim_ratio: compaction_config::tombstone_reclaim_ratio(),
                 enable_emergency_picker: compaction_config::enable_emergency_picker(),
+                min_keep_alive_time_secs: compaction_config::min_keep_alive_time_secs(),
             },
         }
     }

src/meta/src/hummock/compaction/picker/tombstone_reclaim_compaction_picker.rs

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use risingwave_common::util::epoch::Epoch;
 use risingwave_pb::hummock::hummock_version::Levels;
 use risingwave_pb::hummock::InputLevel;
 
@@ -25,6 +26,7 @@ pub struct TombstoneReclaimCompactionPicker {
     overlap_strategy: Arc<dyn OverlapStrategy>,
     delete_ratio: u64,
     range_delete_ratio: u64,
+    min_keep_alive_time_secs: u64,
 }
 
 #[derive(Default)]
@@ -37,11 +39,13 @@ impl TombstoneReclaimCompactionPicker {
         overlap_strategy: Arc<dyn OverlapStrategy>,
         delete_ratio: u64,
         range_delete_ratio: u64,
+        min_keep_alive_time_secs: u64,
     ) -> Self {
         Self {
             overlap_strategy,
             delete_ratio,
             range_delete_ratio,
+            min_keep_alive_time_secs,
         }
     }
 
@@ -55,14 +59,18 @@ impl TombstoneReclaimCompactionPicker {
         if state.last_level == 0 {
             state.last_level = 1;
         }
+        let current_time = Epoch::physical_now() / 1000;
 
         while state.last_level <= levels.levels.len() {
             let mut select_input_ssts = vec![];
             for sst in &levels.levels[state.last_level - 1].table_infos {
                 let need_reclaim = (sst.range_tombstone_count * 100
                     >= sst.total_key_count * self.range_delete_ratio)
                     || (sst.stale_key_count * 100 >= sst.total_key_count * self.delete_ratio);
-                if !need_reclaim || level_handlers[state.last_level].is_pending_compact(&sst.sst_id)
+                let shall_keep = sst.create_time + self.min_keep_alive_time_secs > current_time;
+                if !need_reclaim
+                    || shall_keep
+                    || level_handlers[state.last_level].is_pending_compact(&sst.sst_id)
                 {
                     continue;
                 }

src/meta/src/hummock/manager/mod.rs

@@ -1406,6 +1406,15 @@ impl HummockManager {
         let versioning = versioning_guard.deref_mut();
         let _timer = start_measure_real_process_timer!(self);
 
+        let task_end_time = Epoch::physical_now() / 1000;
+        {
+            // apply result
+            compact_task.set_task_status(task_status);
+            compact_task.sorted_output_ssts = sorted_output_ssts;
+            for sst in &mut compact_task.sorted_output_ssts {
+                sst.create_time = task_end_time;
+            }
+        }
         let mut current_version = versioning.current_version.clone();
         // purge stale compact_status
         for group_id in original_keys {

src/storage/src/hummock/sstable/builder.rs

@@ -511,6 +511,7 @@ impl<W: SstableWriter, F: FilterBuilder> SstableBuilder<W, F> {
             uncompressed_file_size: uncompressed_file_size + meta.encoded_size() as u64,
             min_epoch: cmp::min(min_epoch, tombstone_min_epoch),
             max_epoch: cmp::max(max_epoch, tombstone_max_epoch),
+            create_time: 0,
             range_tombstone_count: meta.monotonic_tombstone_events.len() as u64,
         };
         tracing::trace!(