feat: new table option data_retention_period_in_hours (#16266)

* feat: new table option `data_retention_period_in_hours`

* refactor: rename things

* add logic test

* adjust logic test

Author: dantengsky

src/common/io/src/constants.rs

@@ -32,3 +32,6 @@ pub const DEFAULT_BLOCK_INDEX_BUFFER_SIZE: usize = 300 * 1024;
 pub const DEFAULT_BLOCK_MAX_ROWS: usize = 1000 * 1000;
 // The min number of a block by default.
 pub const DEFAULT_BLOCK_MIN_ROWS: usize = 800 * 1000;
+
+// The min values of table option data_retention_period_in_hours
+pub const DEFAULT_MIN_TABLE_LEVEL_DATA_RETENTION_PERIOD_IN_HOURS: u64 = 1;

src/query/ee/src/storages/fuse/operations/vacuum_table.rs

@@ -17,7 +17,6 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use chrono::DateTime;
-use chrono::Duration;
 use chrono::Utc;
 use databend_common_catalog::table::NavigationPoint;
 use databend_common_catalog::table::Table;
@@ -355,7 +354,7 @@ pub async fn do_vacuum(
         .await?;
     let status = format!("do_vacuum: purged table, cost:{:?}", start.elapsed());
     ctx.set_status_info(&status);
-    let retention = Duration::days(ctx.get_settings().get_data_retention_time_in_days()? as i64);
+    let retention = fuse_table.get_data_retention_period(ctx.as_ref())?;
     // use min(now - get_retention_period(), retention_time) as gc orphan files retention time
     // to protect files that generated by txn which has not been committed being gc.
     let retention_time = std::cmp::min(chrono::Utc::now() - retention, retention_time);

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

@@ -17,6 +17,7 @@ use std::collections::HashSet;
 use std::sync::Arc;
 use std::sync::LazyLock;
 
+use chrono::Duration;
 use chrono::Utc;
 use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_config::GlobalConfig;
@@ -26,6 +27,7 @@ use databend_common_expression::is_internal_column;
 use databend_common_expression::TableSchemaRef;
 use databend_common_expression::TableSchemaRefExt;
 use databend_common_io::constants::DEFAULT_BLOCK_MAX_ROWS;
+use databend_common_io::constants::DEFAULT_MIN_TABLE_LEVEL_DATA_RETENTION_PERIOD_IN_HOURS;
 use databend_common_license::license::Feature;
 use databend_common_license::license::Feature::ComputedColumn;
 use databend_common_license::license::Feature::InvertedIndex;
@@ -42,12 +44,14 @@ use databend_common_meta_app::schema::TableNameIdent;
 use databend_common_meta_app::schema::TableStatistics;
 use databend_common_meta_types::MatchSeq;
 use databend_common_pipeline_core::ExecutionInfo;
+use databend_common_settings::Settings;
 use databend_common_sql::field_default_value;
 use databend_common_sql::plans::CreateTablePlan;
 use databend_common_sql::BloomIndexColumns;
 use databend_common_storages_fuse::io::MetaReaders;
 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_PERIOD_IN_HOURS;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_PER_BLOCK;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ROW_PER_PAGE;
@@ -448,6 +452,8 @@ impl CreateTableInterpreter {
         is_valid_change_tracking(&table_meta.options)?;
         // check random seed
         is_valid_random_seed(&table_meta.options)?;
+        // check table level data_retention_period_in_hours
+        is_valid_data_retention_period(&table_meta.options)?;
 
         for table_option in table_meta.options.iter() {
             let key = table_option.0.to_lowercase();
@@ -498,6 +504,7 @@ pub static CREATE_TABLE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new
     r.insert(FUSE_OPT_KEY_ROW_PER_BLOCK);
     r.insert(FUSE_OPT_KEY_BLOCK_IN_MEM_SIZE_THRESHOLD);
     r.insert(FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD);
+    r.insert(FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS);
 
     r.insert(OPT_KEY_BLOOM_INDEX_COLUMNS);
     r.insert(OPT_KEY_TABLE_COMPRESSION);
@@ -561,6 +568,32 @@ pub fn is_valid_row_per_block(options: &BTreeMap<String, String>) -> Result<()>
     Ok(())
 }
 
+pub fn is_valid_data_retention_period(options: &BTreeMap<String, String>) -> Result<()> {
+    if let Some(value) = options.get(FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS) {
+        let new_duration_in_hours = value.parse::<u64>()?;
+
+        if new_duration_in_hours < DEFAULT_MIN_TABLE_LEVEL_DATA_RETENTION_PERIOD_IN_HOURS {
+            return Err(ErrorCode::TableOptionInvalid(format!(
+                "Invalid data_retention_period_in_hours {:?}, it should not be lesser than {:?}",
+                new_duration_in_hours, DEFAULT_MIN_TABLE_LEVEL_DATA_RETENTION_PERIOD_IN_HOURS
+            )));
+        }
+
+        let default_max_period_in_days = Settings::get_max_data_retention_period_in_days();
+
+        let default_max_duration = Duration::days(default_max_period_in_days as i64);
+        let new_duration = Duration::hours(new_duration_in_hours as i64);
+
+        if new_duration > default_max_duration {
+            return Err(ErrorCode::TableOptionInvalid(format!(
+                "Invalid data_retention_period_in_hours {:?}, it should not be larger than {:?}",
+                new_duration, default_max_duration
+            )));
+        }
+    }
+    Ok(())
+}
+
 pub fn is_valid_bloom_index_columns(
     options: &BTreeMap<String, String>,
     schema: TableSchemaRef,

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

@@ -32,6 +32,7 @@ use log::error;
 use super::interpreter_table_create::is_valid_block_per_segment;
 use super::interpreter_table_create::is_valid_bloom_index_columns;
 use super::interpreter_table_create::is_valid_create_opt;
+use super::interpreter_table_create::is_valid_data_retention_period;
 use super::interpreter_table_create::is_valid_row_per_block;
 use crate::interpreters::Interpreter;
 use crate::pipelines::PipelineBuildResult;
@@ -66,6 +67,9 @@ impl Interpreter for SetOptionsInterpreter {
         is_valid_block_per_segment(&self.plan.set_options)?;
         // check row_per_block
         is_valid_row_per_block(&self.plan.set_options)?;
+        // check row_per_block
+        is_valid_data_retention_period(&self.plan.set_options)?;
+
         // check storage_format
         let error_str = "invalid opt for fuse table in alter table statement";
         if self.plan.set_options.contains_key(OPT_KEY_STORAGE_FORMAT) {

src/query/service/src/interpreters/interpreter_table_vacuum.rs

@@ -14,7 +14,6 @@
 
 use std::sync::Arc;
 
-use chrono::Duration;
 use databend_common_catalog::table::TableExt;
 use databend_common_exception::Result;
 use databend_common_expression::types::StringType;
@@ -124,12 +123,12 @@ impl Interpreter for VacuumTableInterpreter {
         // check mutability
         table.check_mutable()?;
 
-        let duration = Duration::days(ctx.get_settings().get_data_retention_time_in_days()? as i64);
+        let fuse_table = FuseTable::try_from_table(table.as_ref())?;
+        let duration = fuse_table.get_data_retention_period(ctx.as_ref())?;
 
         let retention_time = chrono::Utc::now() - duration;
         let ctx = self.ctx.clone();
 
-        let fuse_table = FuseTable::try_from_table(table.as_ref())?;
         let handler = get_vacuum_handler();
         let purge_files_opt = handler
             .do_vacuum(

src/query/settings/src/settings_default.rs

@@ -847,7 +847,7 @@ impl DefaultSettings {
     /// The maximum number of days that data can be retained.
     /// The max is read from the global config:data_retention_time_in_days_max
     /// If the global config is not set, the default value is 90 days.
-    fn data_retention_time_in_days_max() -> u64 {
+    pub(crate) fn data_retention_time_in_days_max() -> u64 {
         match GlobalConfig::try_get_instance() {
             None => 90,
             Some(conf) => conf.query.data_retention_time_in_days_max,

src/query/settings/src/settings_getter_setter.rs

@@ -681,4 +681,8 @@ impl Settings {
     pub fn get_enable_last_snapshot_location_hint(&self) -> Result<bool> {
         Ok(self.try_get_u64("enable_last_snapshot_location_hint")? == 1)
     }
+
+    pub fn get_max_data_retention_period_in_days() -> u64 {
+        DefaultSettings::data_retention_time_in_days_max()
+    }
 }

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

@@ -18,6 +18,8 @@ pub const FUSE_OPT_KEY_ROW_PER_BLOCK: &str = "row_per_block";
 pub const FUSE_OPT_KEY_ROW_PER_PAGE: &str = "row_per_page";
 pub const FUSE_OPT_KEY_ROW_AVG_DEPTH_THRESHOLD: &str = "row_avg_depth_threshold";
 
+pub const FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS: &str = "data_retention_period_in_hours";
+
 pub const FUSE_TBL_BLOCK_PREFIX: &str = "_b";
 pub const FUSE_TBL_BLOCK_INDEX_PREFIX: &str = "_i";
 pub const FUSE_TBL_XOR_BLOOM_INDEX_PREFIX: &str = "_i_b_v2";

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

@@ -21,6 +21,8 @@ use std::str;
 use std::str::FromStr;
 use std::sync::Arc;
 
+use chrono::Duration;
+use chrono::TimeDelta;
 use databend_common_catalog::catalog::StorageDescription;
 use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::plan::PartStatistics;
@@ -109,6 +111,7 @@ use crate::DEFAULT_ROW_PER_PAGE;
 use crate::DEFAULT_ROW_PER_PAGE_FOR_BLOCKING;
 use crate::FUSE_OPT_KEY_BLOCK_IN_MEM_SIZE_THRESHOLD;
 use crate::FUSE_OPT_KEY_BLOCK_PER_SEGMENT;
+use crate::FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS;
 use crate::FUSE_OPT_KEY_ROW_PER_BLOCK;
 use crate::FUSE_OPT_KEY_ROW_PER_PAGE;
 use crate::FUSE_TBL_LAST_SNAPSHOT_HINT;
@@ -438,7 +441,7 @@ impl FuseTable {
         })
     }
 
-    pub fn transient(&self) -> bool {
+    pub fn is_transient(&self) -> bool {
         self.table_info.meta.options.contains_key("TRANSIENT")
     }
 
@@ -476,6 +479,21 @@ impl FuseTable {
             .map(|v| v.data_type().clone())
             .collect()
     }
+
+    pub fn get_data_retention_period(&self, ctx: &dyn TableContext) -> Result<TimeDelta> {
+        let retention_period = if let Some(v) = self
+            .table_info
+            .meta
+            .options
+            .get(FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS)
+        {
+            let retention_period = v.parse::<u64>()?;
+            Duration::hours(retention_period as i64)
+        } else {
+            Duration::days(ctx.get_settings().get_data_retention_time_in_days()? as i64)
+        };
+        Ok(retention_period)
+    }
 }
 
 #[async_trait::async_trait]

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

@@ -177,7 +177,7 @@ where F: SnapshotGenerator + Send + 'static
     }
 
     fn do_purge(table: &FuseTable, snapshot_gen: &F) -> bool {
-        if table.transient() {
+        if table.is_transient() {
             return true;
         }