Repository: risingwavelabs/risingwave (language: python)

Issue:
refactor(meta): proactively GC more stale objects.
I hereby agree to the terms of the [RisingWave Labs, Inc. Contributor License Agreement](https://gist.github.com/TennyZhuang/f00be7f16996ea48effb049aa7be4d66#file-rw_cla).

## What's changed and what's your intention?

Previouly all SST objects are retained for at least `min_sst_retention_time_sec` even they're indeed useless.
This PR proactively GC these objects to reduce storage overhead.

## Checklist

- [ ] I have written necessary rustdoc comments
- [ ] I have added necessary unit tests and integration tests
- [ ] I have added test labels as necessary. See [details](https://github.com/risingwavelabs/risingwave/blob/main/docs/developer-guide.md#ci-labels-guide).
- [ ] I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features #7934).
- [ ] My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
- [ ] All checks passed in `./risedev check` (or alias, `./risedev c`)
- [ ] My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)
<!-- To manually trigger a benchmark, please check out [Notion](https://www.notion.so/risingwave-labs/Manually-trigger-nexmark-performance-dashboard-test-b784f1eae1cf48889b2645d020b6b7d3). -->
- [ ] My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the [details](https://github.com/risingwavelabs/risingwave/blob/main/CONTRIBUTING.md))

## Documentation

- [ ] My PR needs documentation updates. (Please use the **Release note** section below to summarize the impact on users)

## Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.


<!--
Please create a release note for your changes.

Discuss technical details in the "What's changed" section, and
focus on the impact on users in the release note.

You should also mention the environment or conditions where the impact may occur.
-->

</details>

cherry-pick refactor(meta): proactively GC more stale objects. (#19474) to branch release-2.1
Cherry picking \#19474 onto branch release-2.1,

This PR/issue was created by cherry-pick action from commit ad1d929a3298a4825c548117db4b48e425005de1.,
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:

Fix patch:
diff --git a/src/common/src/config.rs b/src/common/src/config.rs
index 06b37179f31d3..be8c3dc46de9f 100644
--- a/src/common/src/config.rs
+++ b/src/common/src/config.rs
@@ -1372,7 +1372,7 @@ pub mod default {
         use crate::config::{DefaultParallelism, MetaBackend};
 
         pub fn min_sst_retention_time_sec() -> u64 {
-            3600 * 3
+            3600 * 6
         }
 
         pub fn gc_history_retention_time_sec() -> u64 {
@@ -1380,7 +1380,7 @@ pub mod default {
         }
 
         pub fn full_gc_interval_sec() -> u64 {
-            600
+            3600
         }
 
         pub fn full_gc_object_limit() -> u64 {
diff --git a/src/common/src/system_param/mod.rs b/src/common/src/system_param/mod.rs
index 4d3a994631534..aa305c8d1aa84 100644
--- a/src/common/src/system_param/mod.rs
+++ b/src/common/src/system_param/mod.rs
@@ -422,7 +422,7 @@ impl ValidateOnSet for OverrideValidateOnSet {
 
     fn time_travel_retention_ms(v: &u64) -> Result<()> {
         // This is intended to guarantee that non-time-travel batch query can still function even compute node's recent versions doesn't include the desired version.
-        let min_retention_ms = 600_000;
+        let min_retention_ms = 60_000;
         if *v < min_retention_ms {
             return Err(format!(
                 "time_travel_retention_ms cannot be less than {min_retention_ms}"
diff --git a/src/config/docs.md b/src/config/docs.md
index a349bf98cc21a..0eaee5db1cf03 100644
--- a/src/config/docs.md
+++ b/src/config/docs.md
@@ -37,7 +37,7 @@ This page is automatically generated by `./risedev generate-example-config`
 | enable_hummock_data_archive | If enabled, `SSTable` object file and version delta will be retained. `SSTable` object file need to be deleted via full GC. version delta need to be manually deleted. | false |
 | event_log_channel_max_size | Keeps the latest N events per channel. | 10 |
 | event_log_enabled |  | true |
-| full_gc_interval_sec | Interval of automatic hummock full GC. | 600 |
+| full_gc_interval_sec | Interval of automatic hummock full GC. | 3600 |
 | full_gc_object_limit | Max number of object per full GC job can fetch. | 100000 |
 | gc_history_retention_time_sec | Duration in seconds to retain garbage collection history data. | 21600 |
 | hummock_time_travel_snapshot_interval | The interval at which a Hummock version snapshot is taken for time travel. Larger value indicates less storage overhead but worse query performance. | 100 |
@@ -47,7 +47,7 @@ This page is automatically generated by `./risedev generate-example-config`
 | max_inflight_time_travel_query | Max number of inflight time travel query. | 1000 |
 | meta_leader_lease_secs |  | 30 |
 | min_delta_log_num_for_hummock_version_checkpoint | The minimum delta log number a new checkpoint should compact, otherwise the checkpoint attempt is rejected. | 10 |
-| min_sst_retention_time_sec | Objects within `min_sst_retention_time_sec` won't be deleted by hummock full GC, even they are dangling. | 10800 |
+| min_sst_retention_time_sec | Objects within `min_sst_retention_time_sec` won't be deleted by hummock full GC, even they are dangling. | 21600 |
 | min_table_split_write_throughput | If the size of one table is smaller than `min_table_split_write_throughput`, we would not split it to an single group. | 4194304 |
 | move_table_size_limit |  | 10737418240 |
 | node_num_monitor_interval_sec |  | 10 |
diff --git a/src/config/example.toml b/src/config/example.toml
index 6aec9f8d70f83..a41ab0755d91c 100644
--- a/src/config/example.toml
+++ b/src/config/example.toml
@@ -14,8 +14,8 @@ threshold_auto = 0.8999999761581421
 dir = "./"
 
 [meta]
-min_sst_retention_time_sec = 10800
-full_gc_interval_sec = 600
+min_sst_retention_time_sec = 21600
+full_gc_interval_sec = 3600
 full_gc_object_limit = 100000
 gc_history_retention_time_sec = 21600
 max_inflight_time_travel_query = 1000
diff --git a/src/meta/model/migration/src/lib.rs b/src/meta/model/migration/src/lib.rs
index 4ae39aea29198..f20278e225ab1 100644
--- a/src/meta/model/migration/src/lib.rs
+++ b/src/meta/model/migration/src/lib.rs
@@ -23,6 +23,7 @@ mod m20240806_143329_add_rate_limit_to_source_catalog;
 mod m20240820_081248_add_time_travel_per_table_epoch;
 mod m20240911_083152_variable_vnode_count;
 mod m20241016_065621_hummock_gc_history;
+mod m20241120_182555_hummock_add_time_travel_sst_index;
 
 pub struct Migrator;
 
@@ -83,6 +84,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20240820_081248_add_time_travel_per_table_epoch::Migration),
             Box::new(m20240911_083152_variable_vnode_count::Migration),
             Box::new(m20241016_065621_hummock_gc_history::Migration),
+            Box::new(m20241120_182555_hummock_add_time_travel_sst_index::Migration),
         ]
     }
 }
diff --git a/src/meta/model/migration/src/m20240701_060504_hummock_time_travel.rs b/src/meta/model/migration/src/m20240701_060504_hummock_time_travel.rs
index 7dec44913dc8f..d831122c79bdf 100644
--- a/src/meta/model/migration/src/m20240701_060504_hummock_time_travel.rs
+++ b/src/meta/model/migration/src/m20240701_060504_hummock_time_travel.rs
@@ -109,7 +109,7 @@ impl MigrationTrait for Migration {
 }
 
 #[derive(DeriveIden)]
-enum HummockSstableInfo {
+pub(crate) enum HummockSstableInfo {
     Table,
     SstId,
     ObjectId,
diff --git a/src/meta/model/migration/src/m20241120_182555_hummock_add_time_travel_sst_index.rs b/src/meta/model/migration/src/m20241120_182555_hummock_add_time_travel_sst_index.rs
new file mode 100644
index 0000000000000..b948dd674ec15
--- /dev/null
+++ b/src/meta/model/migration/src/m20241120_182555_hummock_add_time_travel_sst_index.rs
@@ -0,0 +1,36 @@
+use sea_orm_migration::prelude::*;
+
+use crate::m20240701_060504_hummock_time_travel::HummockSstableInfo;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+const IDX_HUMMOCK_SSTABLE_INFO_OBJECT_ID: &str = "idx_hummock_sstable_info_object_id";
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_index(
+                Index::create()
+                    .table(HummockSstableInfo::Table)
+                    .name(IDX_HUMMOCK_SSTABLE_INFO_OBJECT_ID)
+                    .col(HummockSstableInfo::ObjectId)
+                    .to_owned(),
+            )
+            .await?;
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .drop_index(
+                Index::drop()
+                    .table(HummockSstableInfo::Table)
+                    .name(IDX_HUMMOCK_SSTABLE_INFO_OBJECT_ID)
+                    .to_owned(),
+            )
+            .await?;
+        Ok(())
+    }
+}
diff --git a/src/meta/node/src/server.rs b/src/meta/node/src/server.rs
index 73c07ccf82599..626ac32a8171e 100644
--- a/src/meta/node/src/server.rs
+++ b/src/meta/node/src/server.rs
@@ -589,7 +589,7 @@ pub async fn start_service_as_election_leader(
     // sub_tasks executed concurrently. Can be shutdown via shutdown_all
     sub_tasks.extend(hummock::start_hummock_workers(
         hummock_manager.clone(),
-        // compaction_scheduler,
+        backup_manager.clone(),
         &env.opts,
     ));
     sub_tasks.push(start_worker_info_monitor(
diff --git a/src/meta/src/hummock/manager/checkpoint.rs b/src/meta/src/hummock/manager/checkpoint.rs
index 3949331098441..12f831fca3c67 100644
--- a/src/meta/src/hummock/manager/checkpoint.rs
+++ b/src/meta/src/hummock/manager/checkpoint.rs
@@ -220,8 +220,17 @@ impl HummockManager {
                     .collect(),
             });
         }
-        // We can directly discard reference to stale objects that will no longer be used.
         let min_pinned_version_id = self.context_info.read().await.min_pinned_version_id();
+        let may_delete_object = stale_objects
+            .iter()
+            .filter_map(|(version_id, object_ids)| {
+                if *version_id >= min_pinned_version_id {
+                    return None;
+                }
+                Some(object_ids.id.clone())
+            })
+            .flatten();
+        self.gc_manager.add_may_delete_object_ids(may_delete_object);
         stale_objects.retain(|version_id, _| *version_id >= min_pinned_version_id);
         let new_checkpoint = HummockVersionCheckpoint {
             version: current_version.clone(),
diff --git a/src/meta/src/hummock/manager/gc.rs b/src/meta/src/hummock/manager/gc.rs
index 12619cae19773..e2308927fff3a 100644
--- a/src/meta/src/hummock/manager/gc.rs
+++ b/src/meta/src/hummock/manager/gc.rs
@@ -48,6 +48,8 @@ pub(crate) struct GcManager {
     store: ObjectStoreRef,
     path_prefix: String,
     use_new_object_prefix_strategy: bool,
+    /// These objects may still be used by backup or time travel.
+    may_delete_object_ids: parking_lot::Mutex<HashSet<HummockSstableObjectId>>,
 }
 
 impl GcManager {
@@ -60,6 +62,7 @@ impl GcManager {
             store,
             path_prefix: path_prefix.to_owned(),
             use_new_object_prefix_strategy,
+            may_delete_object_ids: Default::default(),
         }
     }
 
@@ -101,7 +104,7 @@ impl GcManager {
         prefix: Option<String>,
         start_after: Option<String>,
         limit: Option<u64>,
-    ) -> Result<(Vec<HummockSstableObjectId>, u64, u64, Option<String>)> {
+    ) -> Result<(HashSet<HummockSstableObjectId>, u64, u64, Option<String>)> {
         tracing::debug!(
             sst_retention_watermark,
             prefix,
@@ -139,7 +142,7 @@ impl GcManager {
                 };
                 async move { result }
             })
-            .try_collect::<Vec<HummockSstableObjectId>>()
+            .try_collect::<HashSet<HummockSstableObjectId>>()
             .await?;
         Ok((
             filtered,
@@ -148,6 +151,28 @@ impl GcManager {
             next_start_after,
         ))
     }
+
+    pub fn add_may_delete_object_ids(
+        &self,
+        may_delete_object_ids: impl Iterator<Item = HummockSstableObjectId>,
+    ) {
+        self.may_delete_object_ids
+            .lock()
+            .extend(may_delete_object_ids);
+    }
+
+    /// Takes if `least_count` elements available.
+    pub fn try_take_may_delete_object_ids(
+        &self,
+        least_count: usize,
+    ) -> Option<HashSet<HummockSstableObjectId>> {
+        let mut guard = self.may_delete_object_ids.lock();
+        if guard.len() < least_count {
+            None
+        } else {
+            Some(std::mem::take(&mut *guard))
+        }
+    }
 }
 
 impl HummockManager {
@@ -295,6 +320,15 @@ impl HummockManager {
         tracing::info!(total_object_count, total_object_size, "Finish GC");
         self.metrics.total_object_size.set(total_object_size as _);
         self.metrics.total_object_count.set(total_object_count as _);
+        match self.time_travel_pinned_object_count().await {
+            Ok(count) => {
+                self.metrics.time_travel_object_count.set(count as _);
+            }
+            Err(err) => {
+                use thiserror_ext::AsReport;
+                tracing::warn!(error = %err.as_report(), "Failed to count time travel objects.");
+            }
+        }
         Ok(())
     }
 
@@ -302,7 +336,7 @@ impl HummockManager {
     /// Returns number of SSTs to delete.
     pub(crate) async fn complete_gc_batch(
         &self,
-        object_ids: Vec<HummockSstableObjectId>,
+        object_ids: HashSet<HummockSstableObjectId>,
         backup_manager: Option<BackupManagerRef>,
     ) -> Result<usize> {
         if object_ids.is_empty() {
@@ -324,31 +358,23 @@ impl HummockManager {
         metrics
             .full_gc_candidate_object_count
             .observe(candidate_object_number as _);
-        let pinned_object_ids = self
-            .all_object_ids_in_time_travel()
-            .await?
-            .collect::<HashSet<_>>();
-        self.metrics
-            .time_travel_object_count
-            .set(pinned_object_ids.len() as _);
-        // filter by SST id watermark, i.e. minimum id of uncommitted SSTs reported by compute nodes.
+        // filter by metadata backup
         let object_ids = object_ids
             .into_iter()
-            .filter(|id| *id < min_sst_id)
+            .filter(|s| !pinned_by_metadata_backup.contains(s))
             .collect_vec();
-        let after_min_sst_id = object_ids.len();
+        let after_metadata_backup = object_ids.len();
         // filter by time travel archive
-        let object_ids = object_ids
-            .into_iter()
-            .filter(|s| !pinned_object_ids.contains(s))
-            .collect_vec();
+        let object_ids = self
+            .filter_out_objects_by_time_travel(object_ids.into_iter())
+            .await?;
         let after_time_travel = object_ids.len();
-        // filter by metadata backup
+        // filter by SST id watermark, i.e. minimum id of uncommitted SSTs reported by compute nodes.
         let object_ids = object_ids
             .into_iter()
-            .filter(|s| !pinned_by_metadata_backup.contains(s))
+            .filter(|id| *id < min_sst_id)
             .collect_vec();
-        let after_metadata_backup = object_ids.len();
+        let after_min_sst_id = object_ids.len();
         // filter by version
         let after_version = self
             .finalize_objects_to_delete(object_ids.into_iter())
@@ -359,9 +385,9 @@ impl HummockManager {
             .observe(after_version_count as _);
         tracing::info!(
             candidate_object_number,
-            after_min_sst_id,
-            after_time_travel,
             after_metadata_backup,
+            after_time_travel,
+            after_min_sst_id,
             after_version_count,
             "complete gc batch"
         );
@@ -506,6 +532,27 @@ impl HummockManager {
         }
         Ok(total)
     }
+
+    /// Minor GC attempts to delete objects that were part of Hummock version but are no longer in use.
+    pub async fn try_start_minor_gc(&self, backup_manager: BackupManagerRef) -> Result<()> {
+        const MIN_MINOR_GC_OBJECT_COUNT: usize = 1000;
+        let Some(object_ids) = self
+            .gc_manager
+            .try_take_may_delete_object_ids(MIN_MINOR_GC_OBJECT_COUNT)
+        else {
+            return Ok(());
+        };
+        // Objects pinned by either meta backup or time travel should be filtered out.
+        let backup_pinned: HashSet<_> = backup_manager.list_pinned_ssts();
+        let object_ids = object_ids
+            .into_iter()
+            .filter(|s| !backup_pinned.contains(s));
+        let object_ids = self.filter_out_objects_by_time_travel(object_ids).await?;
+        // Retry is not necessary. Full GC will handle these objects eventually.
+        self.delete_objects(object_ids.into_iter().collect())
+            .await?;
+        Ok(())
+    }
 }
 
 async fn collect_min_uncommitted_sst_id(
@@ -585,7 +632,7 @@ mod tests {
 
         // Empty input results immediate return, without waiting heartbeat.
         hummock_manager
-            .complete_gc_batch(vec![], None)
+            .complete_gc_batch(vec![].into_iter().collect(), None)
             .await
             .unwrap();
 
@@ -595,7 +642,9 @@ mod tests {
             3,
             hummock_manager
                 .complete_gc_batch(
-                    vec![i64::MAX as u64 - 2, i64::MAX as u64 - 1, i64::MAX as u64],
+                    vec![i64::MAX as u64 - 2, i64::MAX as u64 - 1, i64::MAX as u64]
+                        .into_iter()
+                        .collect(),
                     None,
                 )
                 .await
@@ -621,7 +670,10 @@ mod tests {
             1,
             hummock_manager
                 .complete_gc_batch(
-                    [committed_object_ids, vec![max_committed_object_id + 1]].concat(),
+                    [committed_object_ids, vec![max_committed_object_id + 1]]
+                        .concat()
+                        .into_iter()
+                        .collect(),
                     None,
                 )
                 .await
diff --git a/src/meta/src/hummock/manager/time_travel.rs b/src/meta/src/hummock/manager/time_travel.rs
index f9d2534fb8c6b..e815cc3ddd3f6 100644
--- a/src/meta/src/hummock/manager/time_travel.rs
+++ b/src/meta/src/hummock/manager/time_travel.rs
@@ -15,7 +15,6 @@
 use std::collections::{HashMap, HashSet, VecDeque};
 
 use anyhow::anyhow;
-use itertools::Itertools;
 use risingwave_common::catalog::TableId;
 use risingwave_common::util::epoch::Epoch;
 use risingwave_hummock_sdk::compaction_group::StaticCompactionGroupId;
@@ -36,8 +35,8 @@ use risingwave_pb::hummock::{PbHummockVersion, PbHummockVersionDelta};
 use sea_orm::sea_query::OnConflict;
 use sea_orm::ActiveValue::Set;
 use sea_orm::{
-    ColumnTrait, Condition, DatabaseTransaction, EntityTrait, QueryFilter, QueryOrder, QuerySelect,
-    TransactionTrait,
+    ColumnTrait, Condition, DatabaseTransaction, EntityTrait, PaginatorTrait, QueryFilter,
+    QueryOrder, QuerySelect, TransactionTrait,
 };
 
 use crate::hummock::error::{Error, Result};
@@ -67,9 +66,9 @@ impl HummockManager {
         &self,
         epoch_watermark: HummockEpoch,
     ) -> Result<()> {
+        let min_pinned_version_id = self.context_info.read().await.min_pinned_version_id();
         let sql_store = self.env.meta_store_ref();
         let txn = sql_store.conn.begin().await?;
-
         let version_watermark = hummock_epoch_to_version::Entity::find()
             .filter(
                 hummock_epoch_to_version::Column::Epoch
@@ -83,6 +82,10 @@ impl HummockManager {
             txn.commit().await?;
             return Ok(());
         };
+        let watermark_version_id = std::cmp::min(
+            version_watermark.version_id,
+            min_pinned_version_id.to_u64().try_into().unwrap(),
+        );
         let res = hummock_epoch_to_version::Entity::delete_many()
             .filter(
                 hummock_epoch_to_version::Column::Epoch
@@ -95,31 +98,35 @@ impl HummockManager {
             "delete {} rows from hummock_epoch_to_version",
             res.rows_affected
         );
-        let earliest_valid_version = hummock_time_travel_version::Entity::find()
-            .filter(
-                hummock_time_travel_version::Column::VersionId.lte(version_watermark.version_id),
-            )
+        let latest_valid_version = hummock_time_travel_version::Entity::find()
+            .filter(hummock_time_travel_version::Column::VersionId.lte(watermark_version_id))
             .order_by_desc(hummock_time_travel_version::Column::VersionId)
             .one(&txn)
             .await?
             .map(|m| HummockVersion::from_persisted_protobuf(&m.version.to_protobuf()));
-        let Some(earliest_valid_version) = earliest_valid_version else {
+        let Some(latest_valid_version) = latest_valid_version else {
             txn.commit().await?;
             return Ok(());
         };
-        let (earliest_valid_version_id, earliest_valid_version_sst_ids) = {
+        let (
+            latest_valid_version_id,
+            latest_valid_version_sst_ids,
+            latest_valid_version_object_ids,
+        ) = {
             (
-                earliest_valid_version.id,
-                earliest_valid_version.get_sst_ids(),
+                latest_valid_version.id,
+                latest_valid_version.get_sst_ids(),
+                latest_valid_version.get_object_ids(),
             )
         };
+        let mut object_ids_to_delete: HashSet<_> = HashSet::default();
         let version_ids_to_delete: Vec<risingwave_meta_model::HummockVersionId> =
             hummock_time_travel_version::Entity::find()
                 .select_only()
                 .column(hummock_time_travel_version::Column::VersionId)
                 .filter(
                     hummock_time_travel_version::Column::VersionId
-                        .lt(earliest_valid_version_id.to_u64()),
+                        .lt(latest_valid_version_id.to_u64()),
                 )
                 .order_by_desc(hummock_time_travel_version::Column::VersionId)
                 .into_tuple()
@@ -131,7 +138,7 @@ impl HummockManager {
                 .column(hummock_time_travel_delta::Column::VersionId)
                 .filter(
                     hummock_time_travel_delta::Column::VersionId
-                        .lt(earliest_valid_version_id.to_u64()),
+                        .lt(latest_valid_version_id.to_u64()),
                 )
                 .into_tuple()
                 .all(&txn)
@@ -146,25 +153,27 @@ impl HummockManager {
                         delta_id_to_delete
                     )))
                 })?;
-            let new_sst_ids = HummockVersionDelta::from_persisted_protobuf(
+            let delta_to_delete = HummockVersionDelta::from_persisted_protobuf(
                 &delta_to_delete.version_delta.to_protobuf(),
-            )
-            .newly_added_sst_ids();
+            );
+            let new_sst_ids = delta_to_delete.newly_added_sst_ids();
             // The SST ids added and then deleted by compaction between the 2 versions.
-            let sst_ids_to_delete = &new_sst_ids - &earliest_valid_version_sst_ids;
+            let sst_ids_to_delete = &new_sst_ids - &latest_valid_version_sst_ids;
             let res = hummock_sstable_info::Entity::delete_many()
                 .filter(hummock_sstable_info::Column::SstId.is_in(sst_ids_to_delete))
                 .exec(&txn)
                 .await?;
+            let new_object_ids = delta_to_delete.newly_added_object_ids();
+            object_ids_to_delete.extend(&new_object_ids - &latest_valid_version_object_ids);
             tracing::debug!(
-                delta_id = delta_to_delete.version_id,
+                delta_id = delta_to_delete.id.to_u64(),
                 "delete {} rows from hummock_sstable_info",
                 res.rows_affected
             );
         }
-        let mut next_version_sst_ids = earliest_valid_version_sst_ids;
+        let mut next_version_sst_ids = latest_valid_version_sst_ids;
         for prev_version_id in version_ids_to_delete {
-            let sst_ids = {
+            let prev_version = {
                 let prev_version = hummock_time_travel_version::Entity::find_by_id(prev_version_id)
                     .one(&txn)
                     .await?
@@ -175,14 +184,16 @@ impl HummockManager {
                         )))
                     })?;
                 HummockVersion::from_persisted_protobuf(&prev_version.version.to_protobuf())
-                    .get_sst_ids()
             };
+            let sst_ids = prev_version.get_sst_ids();
             // The SST ids deleted by compaction between the 2 versions.
             let sst_ids_to_delete = &sst_ids - &next_version_sst_ids;
             let res = hummock_sstable_info::Entity::delete_many()
                 .filter(hummock_sstable_info::Column::SstId.is_in(sst_ids_to_delete))
                 .exec(&txn)
                 .await?;
+            let new_object_ids = prev_version.get_object_ids();
+            object_ids_to_delete.extend(&new_object_ids - &latest_valid_version_object_ids);
             tracing::debug!(
                 prev_version_id,
                 "delete {} rows from hummock_sstable_info",
@@ -190,30 +201,33 @@ impl HummockManager {
             );
             next_version_sst_ids = sst_ids;
         }
+        if !object_ids_to_delete.is_empty() {
+            self.gc_manager
+                .add_may_delete_object_ids(object_ids_to_delete.into_iter());
+        }
 
         let res = hummock_time_travel_version::Entity::delete_many()
             .filter(
-                hummock_time_travel_version::Column::VersionId
-                    .lt(earliest_valid_version_id.to_u64()),
+                hummock_time_travel_version::Column::VersionId.lt(latest_valid_version_id.to_u64()),
             )
             .exec(&txn)
             .await?;
         tracing::debug!(
-            epoch_watermark_version_id = ?version_watermark.version_id,
-            ?earliest_valid_version_id,
+            epoch_watermark_version_id = ?watermark_version_id,
+            ?latest_valid_version_id,
             "delete {} rows from hummock_time_travel_version",
             res.rows_affected
         );
 
         let res = hummock_time_travel_delta::Entity::delete_many()
             .filter(
-                hummock_time_travel_delta::Column::VersionId.lt(earliest_valid_version_id.to_u64()),
+                hummock_time_travel_delta::Column::VersionId.lt(latest_valid_version_id.to_u64()),
             )
             .exec(&txn)
             .await?;
         tracing::debug!(
-            epoch_watermark_version_id = ?version_watermark.version_id,
-            ?earliest_valid_version_id,
+            epoch_watermark_version_id = ?watermark_version_id,
+            ?latest_valid_version_id,
             "delete {} rows from hummock_time_travel_delta",
             res.rows_affected
         );
@@ -222,21 +236,38 @@ impl HummockManager {
         Ok(())
     }
 
-    pub(crate) async fn all_object_ids_in_time_travel(
+    pub(crate) async fn filter_out_objects_by_time_travel(
         &self,
-    ) -> Result<impl Iterator<Item = HummockSstableId>> {
-        let object_ids: Vec<risingwave_meta_model::HummockSstableObjectId> =
-            hummock_sstable_info::Entity::find()
-                .select_only()
-                .column(hummock_sstable_info::Column::ObjectId)
-                .into_tuple()
-                .all(&self.env.meta_store_ref().conn)
-                .await?;
-        let object_ids = object_ids
-            .into_iter()
-            .unique()
-            .map(|object_id| HummockSstableObjectId::try_from(object_id).unwrap());
-        Ok(object_ids)
+        objects: impl Iterator<Item = HummockSstableObjectId>,
+    ) -> Result<HashSet<HummockSstableObjectId>> {
+        // The input object count is much smaller than time travel pinned object count in meta store.
+        // So search input object in meta store.
+        let mut result: HashSet<_> = objects.collect();
+        let mut remain: VecDeque<_> = result.iter().copied().collect();
+        const FILTER_BATCH_SIZE: usize = 1000;
+        while !remain.is_empty() {
+            let batch = remain.drain(..std::cmp::min(remain.len(), FILTER_BATCH_SIZE));
+            let reject_object_ids: Vec<risingwave_meta_model::HummockSstableObjectId> =
+                hummock_sstable_info::Entity::find()
+                    .filter(hummock_sstable_info::Column::ObjectId.is_in(batch))
+                    .select_only()
+                    .column(hummock_sstable_info::Column::ObjectId)
+                    .into_tuple()
+                    .all(&self.env.meta_store_ref().conn)
+                    .await?;
+            for reject in reject_object_ids {
+                let object_id = HummockSstableObjectId::try_from(reject).unwrap();
+                result.remove(&object_id);
+            }
+        }
+        Ok(result)
+    }
+
+    pub(crate) async fn time_travel_pinned_object_count(&self) -> Result<u64> {
+        let count = hummock_sstable_info::Entity::find()
+            .count(&self.env.meta_store_ref().conn)
+            .await?;
+        Ok(count)
     }
 
     /// Attempt to locate the version corresponding to `query_epoch`.
diff --git a/src/meta/src/hummock/mod.rs b/src/meta/src/hummock/mod.rs
index 7b8af79ae960e..cb4be256a4551 100644
--- a/src/meta/src/hummock/mod.rs
+++ b/src/meta/src/hummock/mod.rs
@@ -15,6 +15,7 @@ pub mod compaction;
 pub mod compactor_manager;
 pub mod error;
 mod manager;
+
 pub use manager::*;
 use thiserror_ext::AsReport;
 
@@ -33,11 +34,13 @@ pub use mock_hummock_meta_client::MockHummockMetaClient;
 use tokio::sync::oneshot::Sender;
 use tokio::task::JoinHandle;
 
+use crate::backup_restore::BackupManagerRef;
 use crate::MetaOpts;
 
 /// Start hummock's asynchronous tasks.
 pub fn start_hummock_workers(
     hummock_manager: HummockManagerRef,
+    backup_manager: BackupManagerRef,
     meta_opts: &MetaOpts,
 ) -> Vec<(JoinHandle<()>, Sender<()>)> {
     // These critical tasks are put in their own timer loop deliberately, to avoid long-running ones
@@ -45,6 +48,7 @@ pub fn start_hummock_workers(
     let workers = vec![
         start_checkpoint_loop(
             hummock_manager.clone(),
+            backup_manager,
             Duration::from_secs(meta_opts.hummock_version_checkpoint_interval_sec),
             meta_opts.min_delta_log_num_for_hummock_version_checkpoint,
         ),
@@ -85,6 +89,7 @@ pub fn start_vacuum_metadata_loop(
 
 pub fn start_checkpoint_loop(
     hummock_manager: HummockManagerRef,
+    backup_manager: BackupManagerRef,
     interval: Duration,
     min_delta_log_num: u64,
 ) -> (JoinHandle<()>, Sender<()>) {
@@ -111,7 +116,18 @@ pub fn start_checkpoint_loop(
                 .create_version_checkpoint(min_delta_log_num)
                 .await
             {
-                tracing::warn!(error = %err.as_report(), "Hummock version checkpoint error");
+                tracing::warn!(error = %err.as_report(), "Hummock version checkpoint error.");
+            } else {
+                let backup_manager_2 = backup_manager.clone();
+                let hummock_manager_2 = hummock_manager.clone();
+                tokio::task::spawn(async move {
+                    let _ = hummock_manager_2
+                        .try_start_minor_gc(backup_manager_2)
+                        .await
+                        .inspect_err(|err| {
+                            tracing::warn!(error = %err.as_report(), "Hummock minor GC error.");
+                        });
+                });
             }
         }
     });
diff --git a/src/storage/hummock_sdk/src/time_travel.rs b/src/storage/hummock_sdk/src/time_travel.rs
index 3a58a7daa760c..a767b5caa27d7 100644
--- a/src/storage/hummock_sdk/src/time_travel.rs
+++ b/src/storage/hummock_sdk/src/time_travel.rs
@@ -24,7 +24,7 @@ use crate::sstable_info::SstableInfo;
 use crate::version::{
     HummockVersion, HummockVersionCommon, HummockVersionDelta, HummockVersionDeltaCommon,
 };
-use crate::{CompactionGroupId, HummockSstableId};
+use crate::{CompactionGroupId, HummockSstableId, HummockSstableObjectId};
 
 pub type IncompleteHummockVersion = HummockVersionCommon<SstableIdInVersion>;
 
@@ -167,12 +167,16 @@ impl From<(&HummockVersionDelta, &HashSet<CompactionGroupId>)> for IncompleteHum
     }
 }
 
-pub struct SstableIdInVersion(HummockSstableId);
+pub struct SstableIdInVersion {
+    sst_id: HummockSstableId,
+    object_id: HummockSstableObjectId,
+}
 
 impl From<&SstableIdInVersion> for PbSstableInfo {
     fn from(sst_id: &SstableIdInVersion) -> Self {
         Self {
-            sst_id: sst_id.0,
+            sst_id: sst_id.sst_id,
+            object_id: sst_id.object_id,
             ..Default::default()
         }
     }
@@ -185,8 +189,11 @@ impl From<SstableIdInVersion> for PbSstableInfo {
 }
 
 impl From<&PbSstableInfo> for SstableIdInVersion {
-    fn from(value: &PbSstableInfo) -> Self {
-        SstableIdInVersion(value.sst_id)
+    fn from(s: &PbSstableInfo) -> Self {
+        SstableIdInVersion {
+            sst_id: s.sst_id,
+            object_id: s.object_id,
+        }
     }
 }
 


Test patch:
diff --git a/e2e_test/time_travel/basic.slt b/e2e_test/time_travel/basic.slt
index 6dd65cf7125d3..52b7bca114a7e 100644
--- a/e2e_test/time_travel/basic.slt
+++ b/e2e_test/time_travel/basic.slt
@@ -1,7 +1,7 @@
 statement ok
 SET RW_IMPLICIT_FLUSH TO true;
 
-statement error time_travel_retention_ms cannot be less than 600000
+statement error time_travel_retention_ms cannot be less than 60000
 ALTER SYSTEM SET time_travel_retention_ms to 10;
 
 statement ok