introduce setting retention_period

Author: dantengsky

src/query/service/tests/it/storages/testdata/system-tables.txt

@@ -353,6 +353,7 @@ DB.Table: 'system'.'settings', Table: settings-table_id:1, ver:0, Engine: System
 | max_storage_io_requests         | 64         | 64          | SESSION | The maximum number of concurrent IO requests. By default, it is 64.                                               | UInt64 |
 | prefer_broadcast_join           | 0          | 0           | SESSION | If enable broadcast join, default value: 0                                                                        | UInt64 |
 | quoted_ident_case_sensitive     | 1          | 1           | SESSION | Case sensitivity of quoted identifiers, default value: 1 (aka case-sensitive).                                    | UInt64 |
+| retention_period                | 12         | 12          | SESSION | The retention_period in hours. By default the value is 12 hours.                                                  | UInt64 |
 | row_tag                         | row        | row         | SESSION | In xml format, this field is represented as a row tag, e.g. <row>...</row>.                                       | String |
 | sql_dialect                     | PostgreSQL | PostgreSQL  | SESSION | SQL dialect, support "PostgreSQL" "MySQL" and "Hive", default value: "PostgreSQL".                                | String |
 | storage_read_buffer_size        | 1048576    | 1048576     | SESSION | The size of buffer in bytes for buffered reader of dal. By default, it is 1MB.                                    | UInt64 |

src/query/settings/src/lib.rs

@@ -164,6 +164,15 @@ impl Settings {
                 desc: "The maximum memory usage for processing single query, in bytes. By default the value is determined automatically.",
                 possible_values: None,
             },
+            // retention_period
+            SettingValue {
+                // unit of retention_period is hour
+                default_value: UserSettingValue::UInt64(12),
+                user_setting: UserSetting::create("retention_period", UserSettingValue::UInt64(12)),
+                level: ScopeLevel::Session,
+                desc: "The retention_period in hours. By default the value is 12 hours.",
+                possible_values: None,
+            },
             // max_storage_io_requests
             SettingValue {
                 default_value: UserSettingValue::UInt64(64),
@@ -519,6 +528,16 @@ impl Settings {
         self.try_set_u64(key, val, false)
     }
 
+    pub fn set_retention_period(&self, hours: u64) -> Result<()> {
+        let key = "retention_period";
+        self.try_set_u64(key, hours, false)
+    }
+
+    pub fn get_retention_period(&self) -> Result<u64> {
+        let key = "retention_period";
+        self.try_get_u64(key)
+    }
+
     pub fn get_max_storage_io_requests(&self) -> Result<u64> {
         let key = "max_storage_io_requests";
         self.try_get_u64(key)

src/query/storages/fuse/fuse/src/operations/gc.rs

@@ -39,8 +39,6 @@ use crate::io::SegmentsIO;
 use crate::io::SnapshotsIO;
 use crate::FuseTable;
 
-pub const DEFAULT_RETENTION_PERIOD_HOURS: u32 = 12;
-
 #[derive(Default)]
 struct LocationTuple {
     block_location: HashSet<String>,
@@ -123,9 +121,10 @@ impl FuseTable {
 
             // partition the orphan snapshots by retention interval
             let partitioned_snapshots = Self::apply_retention_rule(
+                ctx.as_ref(),
                 min_snapshot_timestamp,
                 snapshot_lites_extended.orphan_snapshot_lites,
-            );
+            )?;
 
             // filter out segments that still referenced by snapshot that within retention period
             all_segment_locations = Self::filter_out_segments_within_retention(
@@ -359,18 +358,20 @@ impl FuseTable {
     // - those are beyond retention period
     // - those are within retention period
     fn apply_retention_rule(
+        ctx: &dyn TableContext,
         base_timestamp: Option<DateTime<Utc>>,
         snapshot_lites: Vec<(TableSnapshotLite, usize)>,
-    ) -> RetentionPartition {
-        let retention_interval = Duration::hours(DEFAULT_RETENTION_PERIOD_HOURS as i64);
+    ) -> Result<RetentionPartition> {
+        //    let retention_interval = Duration::hours(DEFAULT_RETENTION_PERIOD_HOURS as i64);
+        let retention_interval = Duration::hours(ctx.get_settings().get_retention_period()? as i64);
         let retention_point = base_timestamp.map(|s| s - retention_interval);
         let (beyond_retention, within_retention) = snapshot_lites
             .into_iter()
             .partition(|(lite, _idx)| lite.timestamp < retention_point);
-        RetentionPartition {
+        Ok(RetentionPartition {
             beyond_retention,
             within_retention,
-        }
+        })
     }
 
     // filter out segments that are referenced by orphan snapshots