feat: new table option enable_auto_vacuum (#17884)

* feat: new table option `enable_auto_vacuum`

* cleanup

Author: dantengsky

src/query/service/src/interpreters/common/table_option_validation.rs

@@ -12,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::any::type_name;
 use std::collections::BTreeMap;
 use std::collections::HashSet;
+use std::str::FromStr;
 use std::sync::LazyLock;
 
 use chrono::Duration;
@@ -27,6 +29,7 @@ use databend_common_storages_fuse::FUSE_OPT_KEY_BLOCK_IN_MEM_SIZE_THRESHOLD;
 use databend_common_storages_fuse::FUSE_OPT_KEY_BLOCK_PER_SEGMENT;
 use databend_common_storages_fuse::FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP;
 use databend_common_storages_fuse::FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS;
+use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_AUTO_VACUUM;
 use databend_common_storages_fuse::FUSE_OPT_KEY_FILE_SIZE;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_PER_BLOCK;
@@ -62,6 +65,7 @@ pub static CREATE_FUSE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(
     r.insert(FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD);
     r.insert(FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS);
     r.insert(FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP);
+    r.insert(FUSE_OPT_KEY_ENABLE_AUTO_VACUUM);
 
     r.insert(OPT_KEY_BLOOM_INDEX_COLUMNS);
     r.insert(OPT_KEY_TABLE_COMPRESSION);
@@ -212,17 +216,30 @@ pub fn is_valid_bloom_index_columns(
 pub fn is_valid_change_tracking(
     options: &BTreeMap<String, String>,
 ) -> databend_common_exception::Result<()> {
-    if let Some(value) = options.get(OPT_KEY_CHANGE_TRACKING) {
-        value.to_lowercase().parse::<bool>()?;
-    }
-    Ok(())
+    is_valid_option_of_type::<bool>(options, OPT_KEY_CHANGE_TRACKING)
 }
 
 pub fn is_valid_random_seed(
     options: &BTreeMap<String, String>,
 ) -> databend_common_exception::Result<()> {
-    if let Some(value) = options.get(OPT_KEY_RANDOM_SEED) {
-        value.parse::<u64>()?;
+    is_valid_option_of_type::<u64>(options, OPT_KEY_RANDOM_SEED)
+}
+
+pub fn is_valid_option_of_type<T: FromStr>(
+    options: &BTreeMap<String, String>,
+    option_name: &str,
+) -> databend_common_exception::Result<()>
+where
+    <T as FromStr>::Err: std::fmt::Display,
+{
+    if let Some(value) = options.get(option_name) {
+        value.parse::<T>().map_err(|e| {
+            let msg = format!(
+                "Failed to parse value [{value}] for table option '{option_name}' as type {}: {e}",
+                type_name::<T>(),
+            );
+            ErrorCode::TableOptionInvalid(msg)
+        })?;
     }
     Ok(())
 }

src/query/service/src/interpreters/interpreter_table_create.rs

@@ -48,6 +48,7 @@ use databend_common_sql::DefaultExprBinder;
 use databend_common_storages_fuse::io::MetaReaders;
 use databend_common_storages_fuse::FuseSegmentFormat;
 use databend_common_storages_fuse::FuseStorageFormat;
+use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_AUTO_VACUUM;
 use databend_common_users::RoleCacheManager;
 use databend_common_users::UserApiProvider;
 use databend_enterprise_attach_table::get_attach_table_handler;
@@ -69,6 +70,7 @@ use crate::interpreters::common::table_option_validation::is_valid_bloom_index_c
 use crate::interpreters::common::table_option_validation::is_valid_change_tracking;
 use crate::interpreters::common::table_option_validation::is_valid_create_opt;
 use crate::interpreters::common::table_option_validation::is_valid_data_retention_period;
+use crate::interpreters::common::table_option_validation::is_valid_option_of_type;
 use crate::interpreters::common::table_option_validation::is_valid_random_seed;
 use crate::interpreters::common::table_option_validation::is_valid_row_per_block;
 use crate::interpreters::hook::vacuum_hook::hook_clear_m_cte_temp_table;
@@ -458,6 +460,9 @@ impl CreateTableInterpreter {
         // check table level data_retention_period_in_hours
         is_valid_data_retention_period(&table_meta.options)?;
 
+        // Same as settings of FUSE_OPT_KEY_ENABLE_AUTO_VACUUM, expect value type is unsigned integer
+        is_valid_option_of_type::<u32>(&table_meta.options, FUSE_OPT_KEY_ENABLE_AUTO_VACUUM)?;
+
         for table_option in table_meta.options.iter() {
             let key = table_option.0.to_lowercase();
             if !is_valid_create_opt(&key, &self.plan.engine) {

src/query/service/src/interpreters/interpreter_table_set_options.rs

@@ -32,6 +32,7 @@ use databend_common_storages_fuse::segment_format_from_location;
 use databend_common_storages_fuse::FuseSegmentFormat;
 use databend_common_storages_fuse::FuseTable;
 use databend_common_storages_fuse::TableContext;
+use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_AUTO_VACUUM;
 use databend_storages_common_table_meta::meta::column_oriented_segment::AbstractSegment;
 use databend_storages_common_table_meta::meta::column_oriented_segment::ColumnOrientedSegmentBuilder;
 use databend_storages_common_table_meta::meta::column_oriented_segment::SegmentBuilder;
@@ -52,6 +53,7 @@ use crate::interpreters::common::table_option_validation::is_valid_block_per_seg
 use crate::interpreters::common::table_option_validation::is_valid_bloom_index_columns;
 use crate::interpreters::common::table_option_validation::is_valid_create_opt;
 use crate::interpreters::common::table_option_validation::is_valid_data_retention_period;
+use crate::interpreters::common::table_option_validation::is_valid_option_of_type;
 use crate::interpreters::common::table_option_validation::is_valid_row_per_block;
 use crate::interpreters::Interpreter;
 use crate::pipelines::PipelineBuildResult;
@@ -120,6 +122,10 @@ impl Interpreter for SetOptionsInterpreter {
                 OPT_KEY_CLUSTER_TYPE
             )));
         }
+
+        // Same as settings of FUSE_OPT_KEY_ENABLE_AUTO_VACUUM, expect value type is unsigned integer
+        is_valid_option_of_type::<u32>(&self.plan.set_options, FUSE_OPT_KEY_ENABLE_AUTO_VACUUM)?;
+
         let catalog = self.ctx.get_catalog(self.plan.catalog.as_str()).await?;
         let database = self.plan.database.as_str();
         let table_name = self.plan.table.as_str();

src/query/storages/fuse/src/constants.rs

@@ -22,7 +22,7 @@ pub const FUSE_OPT_KEY_FILE_SIZE: &str = "file_size";
 pub const FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS: &str = "data_retention_period_in_hours";
 pub const FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP: &str =
     "data_retention_num_snapshots_to_keep";
-
+pub const FUSE_OPT_KEY_ENABLE_AUTO_VACUUM: &str = "enable_auto_vacuum";
 pub const FUSE_OPT_KEY_ATTACH_COLUMN_IDS: &str = "attach_column_ids";
 
 pub const FUSE_TBL_BLOCK_PREFIX: &str = "_b";

src/query/storages/fuse/src/operations/common/processors/sink_commit.rs

@@ -59,6 +59,7 @@ use crate::operations::SnapshotGenerator;
 use crate::operations::TransformMergeCommitMeta;
 use crate::operations::TruncateGenerator;
 use crate::FuseTable;
+use crate::FUSE_OPT_KEY_ENABLE_AUTO_VACUUM;
 enum State {
     None,
     FillDefault,
@@ -177,14 +178,31 @@ where F: SnapshotGenerator + Send + Sync + 'static
     ) -> Result<Option<PurgeMode>> {
         let mode = if Self::need_to_purge_all_history(table, snapshot_gen) {
             Some(PurgeMode::PurgeAllHistory)
-        } else if ctx.get_settings().get_enable_auto_vacuum()? {
+        } else if Self::is_auto_vacuum_enabled(ctx, table)? {
             Some(PurgeMode::PurgeAccordingToRetention)
         } else {
             None
         };
         Ok(mode)
     }
 
+    fn is_auto_vacuum_enabled(ctx: &dyn TableContext, table: &FuseTable) -> Result<bool> {
+        // Priority for auto vacuum:
+        // - If table-level option `FUSE_OPT_KEY_ENABLE_AUTO_VACUUM` is set, it takes precedence
+        // - If table-level option is not set, fall back to the setting
+        match table
+            .table_info
+            .options()
+            .get(FUSE_OPT_KEY_ENABLE_AUTO_VACUUM)
+        {
+            Some(v) => {
+                let enabled = v.parse::<u32>()? != 0;
+                Ok(enabled)
+            }
+            None => ctx.get_settings().get_enable_auto_vacuum(),
+        }
+    }
+
     fn is_error_recoverable(&self, e: &ErrorCode) -> bool {
         let code = e.code();
         // When prev_snapshot_id is some, means it is an alter table column modification or truncate.

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

@@ -0,0 +1,82 @@
+## 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 auto_vacuum_tbl_opt;
+
+statement ok
+use auto_vacuum_tbl_opt;
+
+statement ok
+create or replace table t (c int) 'fs:///tmp/tbl_auto_vacuum/' data_retention_num_snapshots_to_keep = 1;
+
+statement ok
+create or replace stage stage_av url = 'fs:///tmp/tbl_auto_vacuum/';
+
+# CASE1: By default, no auto vacuum should be triggered
+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%';
+----
+2
+
+# CASE 2: Table level option should have higher priority than settings
+
+# CASE 2.1: Enable auto_vacuum at table level
+# make sure auto_vacuum is disabled at session setting level
+statement ok
+set enable_auto_vacuum = 0;
+
+statement ok
+alter table t set options(enable_auto_vacuum = 1);
+
+statement ok
+insert into t values(3);
+
+# Auto vacuum should be triggered
+onlyif mysql
+query I
+select count() from list_stage(location=> '@stage_av') where name like '%_ss%';
+----
+1
+
+# CASE 2.1: Disable auto_vacuum at table level
+statement ok
+set enable_auto_vacuum = 1;
+
+statement ok
+alter table t set options(enable_auto_vacuum = 0);
+
+statement ok
+insert into t values(3);
+
+# Auto vacuum should NOT be triggered
+onlyif mysql
+query I
+select count() from list_stage(location=> '@stage_av') where name like '%_ss%';
+----
+2
+
+statement ok
+remove @stage_av;
+
+statement ok
+drop stage stage_av;
+