feat: new settings data_retention_num_snapshots_to_keep (#17893)

feat: new setting `data_retention_num_snapshots_to_keep`

Which specifies how many snapshots to retain during vacuum operations.
- Overrides setting 'data_retention_time_in_days'
- If set to 0, this setting will be ignored

logic test

Author: dantengsky

src/query/settings/src/settings_default.rs

@@ -193,6 +193,13 @@ impl DefaultSettings {
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(0..=data_retention_time_in_days_max)),
                 }),
+                ("data_retention_num_snapshots_to_keep", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(0),
+                    desc: "Specifies how many snapshots to retain during vacuum operations. Overrides 'data_retention_time_in_days'. If set to 0, this setting will be ignored.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(0..=500)),
+                }),
                 ("max_spill_io_requests", DefaultSettingValue {
                     value: UserSettingValue::UInt64(default_max_spill_io_requests),
                     desc: "Sets the maximum number of concurrent spill I/O requests.",

src/query/settings/src/settings_getter_setter.rs

@@ -234,6 +234,10 @@ impl Settings {
         self.try_get_u64("data_retention_time_in_days")
     }
 
+    pub fn get_data_retention_num_snapshots_to_keep(&self) -> Result<u64> {
+        self.try_get_u64("data_retention_num_snapshots_to_keep")
+    }
+
     pub fn get_max_storage_io_requests(&self) -> Result<u64> {
         self.try_get_u64("max_storage_io_requests")
     }

src/query/storages/fuse/src/fuse_table.rs

@@ -463,21 +463,63 @@ impl FuseTable {
         }
     }
 
+    /// Returns the data retention policy for this table.
+    /// Policy is determined in the following priority order:
+    ///   1. Number of snapshots to keep (from table option or setting)
+    ///   2. Time-based retention period (if snapshot count is not specified)
     pub fn get_data_retention_policy(&self, ctx: &dyn TableContext) -> Result<RetentionPolicy> {
-        let table_options = &self.table_info.meta.options;
-
         let policy =
-            if let Some(v) = table_options.get(FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP) {
-                let num_snapshot_keep = v.parse::<usize>()?;
-                RetentionPolicy::ByNumOfSnapshotsToKeep(num_snapshot_keep)
+            // Try to get number of snapshots to keep
+            if let Some(num_snapshots) = self.try_get_policy_by_num_snapshots_to_keep(ctx)? {
+                RetentionPolicy::ByNumOfSnapshotsToKeep(num_snapshots as usize)
             } else {
+                // Fall back to time-based retention policy
                 let duration = self.get_data_retention_period(ctx)?;
                 RetentionPolicy::ByTimePeriod(duration)
             };
 
         Ok(policy)
     }
 
+    /// Tries to retrieve the number of snapshots to keep for the retention policy.
+    /// Priority order:
+    ///   1. Table option (if set to a positive value)
+    ///   2. Global setting (if set to a positive value and table option is not applicable)
+    ///
+    /// Returns Some(value) if a valid positive value is found, None otherwise.
+    fn try_get_policy_by_num_snapshots_to_keep(
+        &self,
+        ctx: &dyn TableContext,
+    ) -> Result<Option<u64>> {
+        // Check table option first (highest priority).
+        //
+        // A positive value means we use this many snapshots for retention.
+        // If value of this table option is not set or is Some(0), we'll check the corresponding setting instead.
+        if let Some(tbl_opt) = self
+            .table_info
+            .meta
+            .options
+            .get(FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP)
+        {
+            let num_snapshots = tbl_opt.parse::<u64>()?;
+            if num_snapshots > 0 {
+                return Ok(Some(num_snapshots));
+            }
+        }
+
+        // Check if there is a valid setting of num snapshots to keep:
+        // Only positive value of setting counts.
+        let settings_value = ctx
+            .get_settings()
+            .get_data_retention_num_snapshots_to_keep()?;
+        if settings_value > 0 {
+            return Ok(Some(settings_value));
+        }
+
+        // No valid num_snapshots_to_keep found
+        Ok(None)
+    }
+
     pub fn get_data_retention_period(&self, ctx: &dyn TableContext) -> Result<Duration> {
         let retention_period = if let Some(v) = self
             .table_info

tests/sqllogictests/suites/ee/03_ee_vacuum/03_0007_setting_num_snapshot_to_keep.test

@@ -0,0 +1,73 @@
+## Copyright 2023 Databend Cloud
+##
+## Licensed under the Elastic License, Version 2.0 (the "License");
+## you may not use this file except in compliance with the License.
+## You may obtain a copy of the License at
+##
+##     https://www.elastic.co/licensing/elastic-license
+##
+## Unless required by applicable law or agreed to in writing, software
+## distributed under the License is distributed on an "AS IS" BASIS,
+## WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+## See the License for the specific language governing permissions and
+## limitations under the License.
+
+statement ok
+create or replace database setting_num_snapshot_to_keep;
+
+statement ok
+use  setting_num_snapshot_to_keep;
+
+statement ok
+create or replace table t (c int) 'fs:///tmp/setting_num_snapshot_to_keep/';
+
+statement ok
+create or replace stage stage_av url = 'fs:///tmp/setting_num_snapshot_to_keep/';
+
+# Enable auto vacuum
+statement ok
+set enable_auto_vacuum = 1;
+
+# CASE1: Setting `data_retention_num_snapshots_to_keep` overrides 'data_retention_time_in_days'
+
+statement ok
+set data_retention_time_in_days = 2;
+
+statement ok
+set data_retention_num_snapshots_to_keep = 1;
+
+statement ok
+insert into t values(1);
+
+statement ok
+insert into t values(2);
+
+onlyif mysql
+query I
+select count() from list_stage(location=> '@stage_av') where name like '%_ss%';
+----
+1
+
+# CASE 2: Set  `data_retention_num_snapshots_to_keep` to 0 will make it effectless
+
+statement ok
+set data_retention_num_snapshots_to_keep = 0;
+
+statement ok
+insert into t values(1);
+
+statement ok
+insert into t values(2);
+
+onlyif mysql
+query I
+select count() from list_stage(location=> '@stage_av') where name like '%_ss%';
+----
+3
+
+statement ok
+remove @stage_av;
+
+statement ok
+drop stage stage_av;
+