From 9e6ec9ff95afc078da19fcb2585c1a393c1003ba Mon Sep 17 00:00:00 2001 From: dantengsky Date: Tue, 1 Sep 2026 11:01:31 +0800 Subject: [PATCH 1/8] fix(storage): avoid collecting vacuum2 result files --- .../src/storages/fuse/operations/handler.rs | 2 +- .../fuse/operations/vacuum_table_v2.rs | 83 +++++-------------- .../it/storages/fuse/operations/vacuum2.rs | 15 +++- .../vacuum_handler/src/vacuum_handler.rs | 4 +- .../fuse_vacuum2/fuse_vacuum2_table.rs | 14 ++-- .../ee/03_ee_vacuum/03_0003_vacuum2.test | 9 +- 6 files changed, 46 insertions(+), 81 deletions(-) diff --git a/src/query/ee/src/storages/fuse/operations/handler.rs b/src/query/ee/src/storages/fuse/operations/handler.rs index 9f26ccaf221..ecff7f62050 100644 --- a/src/query/ee/src/storages/fuse/operations/handler.rs +++ b/src/query/ee/src/storages/fuse/operations/handler.rs @@ -46,7 +46,7 @@ impl VacuumHandler for RealVacuumHandler { table: &dyn Table, ctx: Arc, respect_flash_back: bool, - ) -> Result> { + ) -> Result<()> { do_vacuum2(table, ctx, respect_flash_back).await } diff --git a/src/query/ee/src/storages/fuse/operations/vacuum_table_v2.rs b/src/query/ee/src/storages/fuse/operations/vacuum_table_v2.rs index dc18cdde41a..ce485c087ab 100644 --- a/src/query/ee/src/storages/fuse/operations/vacuum_table_v2.rs +++ b/src/query/ee/src/storages/fuse/operations/vacuum_table_v2.rs @@ -106,7 +106,7 @@ pub async fn do_vacuum2( table: &dyn Table, ctx: Arc, respect_flash_back: bool, -) -> Result> { +) -> Result<()> { let table_info = table.get_table_info(); { if ctx.txn_mgr().lock().is_active() { @@ -114,7 +114,7 @@ pub async fn do_vacuum2( "Transaction is active, skipping vacuum, target table {}", table_info.desc ); - return Ok(vec![]); + return Ok(()); } } @@ -127,7 +127,7 @@ pub async fn do_vacuum2( }) = vacuum_base_snapshot_phase(fuse_table, &ctx, respect_flash_back).await? else { info!("Table {} has no snapshot, stopping vacuum", table_info.desc); - return Ok(vec![]); + return Ok(()); }; let start = std::time::Instant::now(); @@ -232,8 +232,6 @@ pub async fn do_vacuum2( .await?; let inverted_indexes = &table_info.meta.indexes; - let mut removed_files = Vec::new(); - // order is important // indexes should be removed before their blocks, because index locations to gc are generated from block locations. let block_location_prefix = fuse_table.meta_location_generator().block_location_prefix(); @@ -250,7 +248,7 @@ pub async fn do_vacuum2( inverted_indexes, start, }; - let block_gc_stats = purge_blocks_before_gc_root(&block_gc_ctx, &mut removed_files).await?; + let block_gc_stats = purge_blocks_before_gc_root(&block_gc_ctx).await?; ctx.set_status_info(&format!( "Filtered and removed blocks for table {}, elapsed: {:?}, blocks scanned: {}, blocks removed: {}, files removed: {}", table_info.desc, @@ -265,12 +263,10 @@ pub async fn do_vacuum2( // segment stats should be removed before segments. if !stats_to_gc.is_empty() { file_remover.remove_file_in_batch(&stats_to_gc).await?; - removed_files.extend(stats_to_gc.iter().cloned()); } if !segments_to_gc.is_empty() { file_remover.remove_file_in_batch(&segments_to_gc).await?; - removed_files.extend(segments_to_gc.iter().cloned()); } // Evict snapshot caches from the local node. @@ -288,7 +284,6 @@ pub async fn do_vacuum2( } } file_remover.remove_file_in_batch(&snapshots_to_gc).await?; - removed_files.extend(snapshots_to_gc.iter().cloned()); // Legacy branch/tag refs were removed without compatibility guarantees. // Vacuum2 cleans up the old ref snapshot prefix opportunistically, and the @@ -298,32 +293,30 @@ pub async fn do_vacuum2( .ref_snapshot_location_prefix(); let _ = fuse_table.get_operator().remove_all(legacy_ref_dir).await; + let removed_files = block_gc_stats.removed_files + + stats_to_gc.len() + + segments_to_gc.len() + + snapshots_to_gc.len(); ctx.set_status_info(&format!( - "Removed files for table {}, elapsed: {:?}, removed_files: {:?}", + "Removed files for table {}, elapsed: {:?}, files removed: {}", table_info.desc, start.elapsed(), - slice_summary(&removed_files), + removed_files, )); - Ok(removed_files) + Ok(()) } -async fn purge_blocks_before_gc_root( - block_gc: &BlockGcContext<'_>, - removed_files: &mut Vec, -) -> Result { +async fn purge_blocks_before_gc_root(block_gc: &BlockGcContext<'_>) -> Result { info!("Listing block files until prefix: {}", block_gc.until); match block_gc.dal.info().scheme() { - Scheme::Fs => purge_blocks_before_gc_root_fs(block_gc, removed_files).await, - _ => purge_blocks_before_gc_root_object_store_streaming(block_gc, removed_files).await, + Scheme::Fs => purge_blocks_before_gc_root_fs(block_gc).await, + _ => purge_blocks_before_gc_root_object_store_streaming(block_gc).await, } } -async fn purge_blocks_before_gc_root_fs( - block_gc: &BlockGcContext<'_>, - removed_files: &mut Vec, -) -> Result { +async fn purge_blocks_before_gc_root_fs(block_gc: &BlockGcContext<'_>) -> Result { let file_remover = Files::create(Arc::clone(block_gc.ctx), block_gc.dal.clone()); let blocks_before_gc_root = list_gc_candidate_paths_until_prefix_fs( block_gc.dal, @@ -352,27 +345,13 @@ async fn purge_blocks_before_gc_root_fs( if !block_gc.gc_root_blocks.contains(&block_path) { block_chunk.push(block_path); if block_chunk.len() == VACUUM2_BLOCK_DELETE_CHUNK_SIZE { - purge_block_chunk( - &file_remover, - block_gc, - &block_chunk, - removed_files, - &mut stats, - ) - .await?; + purge_block_chunk(&file_remover, block_gc, &block_chunk, &mut stats).await?; block_chunk.clear(); } } } if !block_chunk.is_empty() { - purge_block_chunk( - &file_remover, - block_gc, - &block_chunk, - removed_files, - &mut stats, - ) - .await?; + purge_block_chunk(&file_remover, block_gc, &block_chunk, &mut stats).await?; } Ok(stats) @@ -380,7 +359,6 @@ async fn purge_blocks_before_gc_root_fs( async fn purge_blocks_before_gc_root_object_store_streaming( block_gc: &BlockGcContext<'_>, - removed_files: &mut Vec, ) -> Result { let file_remover = Files::create(Arc::clone(block_gc.ctx), block_gc.dal.clone()); let mut lister = block_gc.dal.lister(block_gc.block_location_prefix).await?; @@ -418,27 +396,13 @@ async fn purge_blocks_before_gc_root_object_store_streaming( block_chunk.push(path.to_owned()); if block_chunk.len() == VACUUM2_BLOCK_DELETE_CHUNK_SIZE { - purge_block_chunk( - &file_remover, - block_gc, - &block_chunk, - removed_files, - &mut stats, - ) - .await?; + purge_block_chunk(&file_remover, block_gc, &block_chunk, &mut stats).await?; block_chunk.clear(); } } if !block_chunk.is_empty() { - purge_block_chunk( - &file_remover, - block_gc, - &block_chunk, - removed_files, - &mut stats, - ) - .await?; + purge_block_chunk(&file_remover, block_gc, &block_chunk, &mut stats).await?; } Ok(stats) @@ -476,7 +440,6 @@ async fn purge_block_chunk( file_remover: &Files, block_gc: &BlockGcContext<'_>, block_chunk: &[String], - removed_files: &mut Vec, stats: &mut BlockGcStats, ) -> Result<()> { if let Err(err) = block_gc.ctx.check_aborting() { @@ -506,13 +469,11 @@ async fn purge_block_chunk( if !indexes_to_gc.is_empty() { file_remover.remove_file_in_batch(&indexes_to_gc).await?; stats.removed_files += indexes_to_gc.len(); - removed_files.extend(indexes_to_gc); } file_remover.remove_file_in_batch(block_chunk).await?; stats.removed_blocks += block_chunk.len(); stats.removed_files += block_chunk.len(); - removed_files.extend(block_chunk.iter().cloned()); block_gc.ctx.set_status_info(&format!( "Removed block chunk for table {}, elapsed: {:?}, block chunk: {}, blocks scanned: {}, blocks removed in chunk: {}, total blocks removed: {}", @@ -713,8 +674,7 @@ mod tests { }; assert_ne!(dal.info().scheme(), Scheme::Fs); - let mut removed_files = Vec::new(); - let stats = purge_blocks_before_gc_root(&block_gc, &mut removed_files).await?; + let stats = purge_blocks_before_gc_root(&block_gc).await?; assert_eq!(stats.scanned_blocks, CANDIDATE_BLOCKS); assert_eq!(stats.removed_blocks, CANDIDATE_BLOCKS - 1); @@ -820,8 +780,7 @@ mod tests { }; anyhow::ensure!(dal.info().scheme() == Scheme::S3, "expected an S3 operator"); - let mut removed_files = Vec::new(); - let stats = purge_blocks_before_gc_root(&block_gc, &mut removed_files).await?; + let stats = purge_blocks_before_gc_root(&block_gc).await?; anyhow::ensure!(stats.scanned_blocks == CANDIDATE_BLOCKS); anyhow::ensure!(stats.removed_blocks == CANDIDATE_BLOCKS - 1); diff --git a/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs b/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs index afa571778e8..de831588cb3 100644 --- a/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs +++ b/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs @@ -220,12 +220,21 @@ async fn test_vacuum2_protected_segments_span_multiple_chunks() -> anyhow::Resul } assert_eq!(live_blocks.len(), SEGMENT_COUNT + 1); - fixture - .execute_command(&format!( - "call system$fuse_vacuum2('{}', '{}')", + let stream = fixture + .execute_query(&format!( + "select * from fuse_vacuum2('{}', '{}')", db_name, tbl_name )) .await?; + let vacuum_result: Vec = stream.try_collect().await?; + assert_eq!( + vacuum_result + .iter() + .map(|block| block.num_rows()) + .sum::(), + 0, + "vacuum2 should not return per-file result rows" + ); // The core assertion: every block still referenced by the live snapshot must // survive. Dropping any protected-segment chunk during the read would leave diff --git a/src/query/ee_features/vacuum_handler/src/vacuum_handler.rs b/src/query/ee_features/vacuum_handler/src/vacuum_handler.rs index 2dec82feaf6..54f24fe251e 100644 --- a/src/query/ee_features/vacuum_handler/src/vacuum_handler.rs +++ b/src/query/ee_features/vacuum_handler/src/vacuum_handler.rs @@ -41,7 +41,7 @@ pub trait VacuumHandler: Sync + Send { table: &dyn Table, ctx: Arc, respect_flash_back: bool, - ) -> Result>; + ) -> Result<()>; async fn do_vacuum_drop_tables( &self, @@ -91,7 +91,7 @@ impl VacuumHandlerWrapper { table: &dyn Table, ctx: Arc, respect_flash_back: bool, - ) -> Result> { + ) -> Result<()> { self.handler .do_vacuum2(table, ctx, respect_flash_back) .await diff --git a/src/query/service/src/table_functions/fuse_vacuum2/fuse_vacuum2_table.rs b/src/query/service/src/table_functions/fuse_vacuum2/fuse_vacuum2_table.rs index 3547bea37b4..c09f554f8ab 100644 --- a/src/query/service/src/table_functions/fuse_vacuum2/fuse_vacuum2_table.rs +++ b/src/query/service/src/table_functions/fuse_vacuum2/fuse_vacuum2_table.rs @@ -23,12 +23,10 @@ use databend_common_catalog::table_args::TableArgs; use databend_common_exception::ErrorCode; use databend_common_exception::Result; use databend_common_expression::DataBlock; -use databend_common_expression::FromData; use databend_common_expression::TableDataType; use databend_common_expression::TableField; use databend_common_expression::TableSchemaRef; use databend_common_expression::TableSchemaRefExt; -use databend_common_expression::types::StringType; use databend_common_license::license::Feature::Vacuum; use databend_common_license::license_manager::LicenseManagerSwitch; use databend_common_storages_fuse::FuseTable; @@ -94,7 +92,7 @@ impl SimpleTableFunc for FuseVacuum2Table { LicenseManagerSwitch::instance().check_enterprise_enabled(ctx.get_license_key(), Vacuum)?; let catalog = ctx.get_catalog(CATALOG_DEFAULT).await?; - let res = match &self.args { + match &self.args { Vacuum2TableArgs::SingleTable { arg_database_name, arg_table_name, @@ -111,9 +109,7 @@ impl SimpleTableFunc for FuseVacuum2Table { } Vacuum2TableArgs::All => self.apply_all_tables(ctx, catalog.as_ref()).await?, }; - Ok(Some(DataBlock::new_from_columns(vec![ - StringType::from_data(res), - ]))) + Ok(None) } fn create(func_name: &str, table_args: TableArgs) -> Result @@ -161,7 +157,7 @@ impl FuseVacuum2Table { database_name: &str, table_name: &str, respect_flash_back: bool, - ) -> Result> { + ) -> Result<()> { let tbl = catalog .get_table(&ctx.get_tenant(), database_name, table_name) .await?; @@ -181,7 +177,7 @@ impl FuseVacuum2Table { &self, ctx: &Arc, catalog: &dyn Catalog, - ) -> Result> { + ) -> Result<()> { let tenant_id = ctx.get_tenant(); let dbs = catalog.list_databases(&tenant_id).await?; let num_db = dbs.len(); @@ -242,6 +238,6 @@ impl FuseVacuum2Table { } } - Ok(vec![]) + Ok(()) } } diff --git a/tests/sqllogictests/suites/ee/03_ee_vacuum/03_0003_vacuum2.test b/tests/sqllogictests/suites/ee/03_ee_vacuum/03_0003_vacuum2.test index 0d4920e1305..de7bd3313c8 100644 --- a/tests/sqllogictests/suites/ee/03_ee_vacuum/03_0003_vacuum2.test +++ b/tests/sqllogictests/suites/ee/03_ee_vacuum/03_0003_vacuum2.test @@ -79,10 +79,11 @@ select count() from list_stage(location=> '@stage_v') where name like '%_ss%'; ---- 4 -# vacuum historical data -# `call system$fuse_vacuum2(...)` also works, but we need to ignore the result -statement ok -select * from fuse_vacuum2('vacuum2', 't') ignore_result; +# vacuum historical data and return no per-file result rows +query I +select count(*) from fuse_vacuum2('vacuum2', 't'); +---- +0 # since retention period is zero, expect only the version generated by # `optimize table .. compact` will be kept From 45761c15f3c235425deed6e4f62d10ef1d95cabf Mon Sep 17 00:00:00 2001 From: dantengsky Date: Tue, 1 Sep 2026 21:51:56 +0800 Subject: [PATCH 2/8] test(storage): update vacuum2 fake-time assertions --- .../00_dummy_cases/00_0002_vacuum2.result | 1 + .../00_dummy_cases/00_0002_vacuum2.sh | 32 ++----------------- ...00_0003_vacuum2_respect_time_travel.result | 1 + .../00_0003_vacuum2_respect_time_travel.sh | 32 ++----------------- 4 files changed, 6 insertions(+), 60 deletions(-) diff --git a/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.result b/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.result index 5f5a2cf6a2f..fb72acf68ee 100644 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.result +++ b/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.result @@ -4,6 +4,7 @@ >>>> select count(*) from fuse_snapshot('default','test_vacuum2') 4 <<<< +>>>> set data_retention_time_in_days = 0;call system$fuse_vacuum2('default','test_vacuum2'); >>>> select count(*) from fuse_snapshot('default','test_vacuum2') 1 <<<< diff --git a/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.sh b/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.sh index 7bae89aa08f..57c8e5d299f 100755 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.sh +++ b/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.sh @@ -3,47 +3,19 @@ CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) . "$CURDIR"/../../../shell_env.sh -SEGMENTS=$(echo "select file_location from fuse_segment('default','test_vacuum2');" | bendsql_connect_root) -BLOCKS=$(echo "select block_location from fuse_block('default','test_vacuum2');" | bendsql_connect_root) - stmt "insert into test_vacuum2 values(2);" -IFS=$'\n' read -d '' -r -a segments <<< "$SEGMENTS" -IFS=$'\n' read -d '' -r -a blocks <<< "$BLOCKS" -blooms=() -for block in "${blocks[@]}"; do - bloom=$(echo "$block" | sed -E 's|(1/[0-9]+)/_b/g([0-9a-f]{32})_v2\.parquet|\1/_i_b_v2/\2_v4.parquet|') - blooms+=("$bloom") -done - stmt "set data_retention_time_in_days = 2;truncate table test_vacuum2;" -SNAPSHOTS=$(echo "select snapshot_location from fuse_snapshot('default','test_vacuum2');" | bendsql_connect_root) -IFS=$'\n' read -d '' -r -a snapshots <<< "$SNAPSHOTS" -to_be_vacuumed=("${snapshots[@]}" "${segments[@]}" "${blocks[@]}" "${blooms[@]}") - # gc root stmt "insert into test_vacuum2 values(3);" - # should have 4 snapshots query "select count(*) from fuse_snapshot('default','test_vacuum2')" -RESULTS=$(echo "set data_retention_time_in_days = 0;select * from fuse_vacuum2('default','test_vacuum2');" | bendsql_connect_root) -IFS=$'\n' read -d '' -r -a results <<< "$RESULTS" - -# verify the vacuum result -sorted_results=($(printf "%s\n" "${results[@]}" | sort)) -sorted_to_be_vacuumed=($(printf "%s\n" "${to_be_vacuumed[@]}" | sort)) - -if [ "$(printf "%s" "${sorted_results[@]}")" != "$(printf "%s" "${sorted_to_be_vacuumed[@]}")" ]; then - echo "Vacuum failed" - echo "Results array: ${sorted_results[@]}" - echo "To be vacuumed array: ${sorted_to_be_vacuumed[@]}" - exit 1 -fi +stmt "set data_retention_time_in_days = 0;call system\$fuse_vacuum2('default','test_vacuum2');" -# remain two snapshots +# only the current snapshot remains query "select count(*) from fuse_snapshot('default','test_vacuum2')" # verify the data diff --git a/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.result b/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.result index 0163305ee37..14aee110822 100644 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.result +++ b/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.result @@ -4,6 +4,7 @@ >>>> select count(*) from fuse_snapshot('default','test_vacuum2_respect_time_travel') 4 <<<< +>>>> set data_retention_time_in_days = 0;call system$fuse_vacuum2('default','test_vacuum2_respect_time_travel',true); >>>> select count(*) from fuse_snapshot('default','test_vacuum2_respect_time_travel') 1 <<<< diff --git a/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.sh b/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.sh index 3e4269ade91..3167a87c2d8 100755 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.sh +++ b/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.sh @@ -3,47 +3,19 @@ CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) . "$CURDIR"/../../../shell_env.sh -SEGMENTS=$(echo "select file_location from fuse_segment('default','test_vacuum2_respect_time_travel');" | bendsql_connect_root) -BLOCKS=$(echo "select block_location from fuse_block('default','test_vacuum2_respect_time_travel');" | bendsql_connect_root) - stmt "insert into test_vacuum2_respect_time_travel values(2);" -IFS=$'\n' read -d '' -r -a segments <<< "$SEGMENTS" -IFS=$'\n' read -d '' -r -a blocks <<< "$BLOCKS" -blooms=() -for block in "${blocks[@]}"; do - bloom=$(echo "$block" | sed -E 's|(1/[0-9]+)/_b/g([0-9a-f]{32})_v2\.parquet|\1/_i_b_v2/\2_v4.parquet|') - blooms+=("$bloom") -done - stmt "set data_retention_time_in_days = 2;truncate table test_vacuum2_respect_time_travel;" -SNAPSHOTS=$(echo "select snapshot_location from fuse_snapshot('default','test_vacuum2_respect_time_travel');" | bendsql_connect_root) -IFS=$'\n' read -d '' -r -a snapshots <<< "$SNAPSHOTS" -to_be_vacuumed=("${snapshots[@]}" "${segments[@]}" "${blocks[@]}" "${blooms[@]}") - # gc root stmt "insert into test_vacuum2_respect_time_travel values(3);" - # should have 4 snapshots query "select count(*) from fuse_snapshot('default','test_vacuum2_respect_time_travel')" -RESULTS=$(echo "set data_retention_time_in_days = 0;select * from fuse_vacuum2('default','test_vacuum2_respect_time_travel',true);" | bendsql_connect_root) -IFS=$'\n' read -d '' -r -a results <<< "$RESULTS" - -# verify the vacuum result -sorted_results=($(printf "%s\n" "${results[@]}" | sort)) -sorted_to_be_vacuumed=($(printf "%s\n" "${to_be_vacuumed[@]}" | sort)) - -if [ "$(printf "%s" "${sorted_results[@]}")" != "$(printf "%s" "${sorted_to_be_vacuumed[@]}")" ]; then - echo "Vacuum failed" - echo "Results array: ${sorted_results[@]}" - echo "To be vacuumed array: ${sorted_to_be_vacuumed[@]}" - exit 1 -fi +stmt "set data_retention_time_in_days = 0;call system\$fuse_vacuum2('default','test_vacuum2_respect_time_travel',true);" -# remain two snapshots +# only the current snapshot remains query "select count(*) from fuse_snapshot('default','test_vacuum2_respect_time_travel')" # verify the data From b1b53c8dbb82d5d7469f12c49a763c1b090a8a79 Mon Sep 17 00:00:00 2001 From: dantengsky Date: Wed, 2 Sep 2026 11:25:31 +0800 Subject: [PATCH 3/8] test(storage): replace fake-time vacuum coverage --- .../action.yml | 15 ---- .github/workflows/reuse.linux.yml | 27 ------- ...run-ee-tests-standalone-fake-time-minio.sh | 42 ----------- .../it/storages/fuse/operations/vacuum2.rs | 70 +++++++++++++++++++ .../storages/fuse/src/operations/vacuum.rs | 67 ++++++++++++++++-- .../00_prepare/00_0000_dummy_prepare.result | 0 .../00_prepare/00_0000_dummy_prepare.sh | 10 --- .../00_prepare/00_0001_vacuum2_prepare.result | 2 - .../00_prepare/00_0001_vacuum2_prepare.sh | 8 --- ...00_0002_vacuum2_respect_time_travel.result | 2 - .../00_0002_vacuum2_respect_time_travel.sh | 8 --- .../00_dummy_cases/00_0001_dummy.result | 1 - .../00_dummy_cases/00_0001_dummy.sh | 29 -------- .../00_dummy_cases/00_0002_vacuum2.result | 14 ---- .../00_dummy_cases/00_0002_vacuum2.sh | 25 ------- ...00_0003_vacuum2_respect_time_travel.result | 14 ---- .../00_0003_vacuum2_respect_time_travel.sh | 25 ------- 17 files changed, 132 insertions(+), 227 deletions(-) delete mode 100644 .github/actions/test_ee_standalone_fake_time_linux/action.yml delete mode 100755 scripts/ci/ci-run-ee-tests-standalone-fake-time-minio.sh delete mode 100755 tests/suites/8_faked_time_prepare/00_prepare/00_0000_dummy_prepare.result delete mode 100755 tests/suites/8_faked_time_prepare/00_prepare/00_0000_dummy_prepare.sh delete mode 100644 tests/suites/8_faked_time_prepare/00_prepare/00_0001_vacuum2_prepare.result delete mode 100755 tests/suites/8_faked_time_prepare/00_prepare/00_0001_vacuum2_prepare.sh delete mode 100644 tests/suites/8_faked_time_prepare/00_prepare/00_0002_vacuum2_respect_time_travel.result delete mode 100755 tests/suites/8_faked_time_prepare/00_prepare/00_0002_vacuum2_respect_time_travel.sh delete mode 100644 tests/suites/9_faked_time/00_dummy_cases/00_0001_dummy.result delete mode 100755 tests/suites/9_faked_time/00_dummy_cases/00_0001_dummy.sh delete mode 100644 tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.result delete mode 100755 tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.sh delete mode 100644 tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.result delete mode 100755 tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.sh diff --git a/.github/actions/test_ee_standalone_fake_time_linux/action.yml b/.github/actions/test_ee_standalone_fake_time_linux/action.yml deleted file mode 100644 index ee99a19af2a..00000000000 --- a/.github/actions/test_ee_standalone_fake_time_linux/action.yml +++ /dev/null @@ -1,15 +0,0 @@ -name: "Test Enterprise features Standalone" -description: "Running stateless tests in standalone mode" -runs: - using: "composite" - steps: - - uses: ./.github/actions/setup_test - - - uses: ./.github/actions/setup_minio - with: - versioning: true - - - name: Run Stateful Tests with Standalone mode - shell: bash - run: | - ./scripts/ci/ci-run-ee-tests-standalone-fake-time-minio.sh diff --git a/.github/workflows/reuse.linux.yml b/.github/workflows/reuse.linux.yml index 5197fc7488e..1b571261d93 100644 --- a/.github/workflows/reuse.linux.yml +++ b/.github/workflows/reuse.linux.yml @@ -510,33 +510,6 @@ jobs: with: name: test-stateful-standalone-linux - # Temporarily commented out, since this job is run on self-hosted, may - # bring troubles to other jobs run on self-hosted. - # - # Maybe we could patch `libfaketime` and let it support jemalloc? - # - # test_ee_standalone_fake_time: - # needs: [build, check] - # runs-on: - # - self-hosted - # - "${{ inputs.runner_arch }}" - # - Linux - # - 2c - # - "${{ inputs.runner_provider }}" - # steps: - # - uses: actions/checkout@v6 - # - uses: ./.github/actions/setup_license - # with: - # runner_provider: ${{ inputs.runner_provider }} - # type: ${{ inputs.license_type }} - # - uses: ./.github/actions/test_ee_standalone_fake_time_linux - # timeout-minutes: 10 - # - name: Upload failure - # if: failure() - # uses: ./.github/actions/artifact_failure - # with: - # name: test-stateful-standalone-fake-time-linux - test_ee_management_mode: needs: [build, check] runs-on: diff --git a/scripts/ci/ci-run-ee-tests-standalone-fake-time-minio.sh b/scripts/ci/ci-run-ee-tests-standalone-fake-time-minio.sh deleted file mode 100755 index cd29960eea2..00000000000 --- a/scripts/ci/ci-run-ee-tests-standalone-fake-time-minio.sh +++ /dev/null @@ -1,42 +0,0 @@ -#!/bin/bash -# Copyright 2020-2021 The Databend Authors. -# SPDX-License-Identifier: Apache-2.0. - -set -e - -echo "*************************************" -echo "* Setting STORAGE_TYPE to S3. *" -echo "* *" -echo "* Please make sure that S3 backend *" -echo "* is ready, and configured properly.*" -echo "*************************************" -export STORAGE_TYPE=s3 -export STORAGE_S3_BUCKET=testbucket -export STORAGE_S3_ROOT=admin -export STORAGE_S3_ENDPOINT_URL=http://127.0.0.1:9900 -export STORAGE_S3_ACCESS_KEY_ID=minioadmin -export STORAGE_S3_SECRET_ACCESS_KEY=minioadmin -export STORAGE_ALLOW_INSECURE=true - -echo "Install dependence" -python3 -m pip install --quiet mysql-connector-python - -echo "Starting standalone DatabendQuery(faked time: 2 days ago)" -sudo date -s "-2 days" -./scripts/ci/deploy/databend-query-standalone.sh - -SCRIPT_PATH="$(cd "$(dirname "$0")" >/dev/null 2>&1 && pwd)" -pushd "$SCRIPT_PATH/../../tests" || exit - -echo "Preparing data (faked time)" -./databend-test --mode 'standalone' --run-dir 8_faked_time_prepare --print-time - -popd -echo "Starting standalone DatabendQuery" -sudo date -s "+2 days" -./scripts/ci/deploy/databend-query-standalone.sh - -pushd "$SCRIPT_PATH/../../tests" || exit - -echo "Testing" -./databend-test --mode 'standalone' --run-dir 9_faked_time --print-time diff --git a/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs b/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs index de831588cb3..12548268510 100644 --- a/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs +++ b/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs @@ -22,10 +22,16 @@ use databend_common_catalog::table::Table; use databend_common_exception::ErrorCode; use databend_common_exception::Result; use databend_common_expression::DataBlock; +use databend_common_meta_app::schema::LeastVisibleTime; +use databend_common_meta_app::schema::least_visible_time_ident::LeastVisibleTimeIdent; use databend_common_storages_fuse::FuseTable; +use databend_common_storages_fuse::io::MetaReaders; use databend_common_storages_fuse::io::SegmentsIO; +use databend_common_storages_fuse::io::SnapshotHistoryReader; +use databend_common_storages_fuse::io::TableMetaLocationGenerator; use databend_enterprise_query::test_kits::context::EESetup; use databend_query::sessions::QueryContext; +use databend_query::sessions::TableContext; use databend_query::sessions::TableContextTableAccess; use databend_query::test_kits::TestFixture; use databend_query::test_kits::execute_command; @@ -129,6 +135,70 @@ async fn test_vacuum2_all() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread")] +async fn test_vacuum2_respect_flash_back_selects_lvt_snapshot() -> anyhow::Result<()> { + let fixture = TestFixture::setup_with_custom(EESetup::new()).await?; + fixture + .default_session() + .get_settings() + .set_data_retention_time_in_days(1)?; + fixture.create_default_database().await?; + + let db_name = fixture.default_db_name(); + let tbl_name = "t_respect_flash_back"; + fixture + .execute_command(&format!("create table {db_name}.{tbl_name} (c int)")) + .await?; + + for value in 1..=3 { + fixture + .execute_command(&format!( + "insert into {db_name}.{tbl_name} values ({value})" + )) + .await?; + tokio::time::sleep(Duration::from_millis(2)).await; + } + + let ctx = fixture.new_query_ctx().await?; + let catalog = ctx.get_default_catalog()?; + let table = catalog + .get_table(&ctx.get_tenant(), &db_name, tbl_name) + .await?; + let fuse_table = FuseTable::try_from_table(table.as_ref())?; + let latest_location = fuse_table.snapshot_loc().unwrap(); + let snapshot_version = TableMetaLocationGenerator::snapshot_version(&latest_location); + let snapshots: Vec<_> = SnapshotHistoryReader::snapshot_history( + MetaReaders::table_snapshot_reader(fuse_table.get_operator()), + latest_location, + snapshot_version, + fuse_table.meta_location_generator().clone(), + ) + .try_collect() + .await?; + assert_eq!(snapshots.len(), 3); + + // History is newest-first. Persist the middle snapshot as LVT so set_lvt() + // returns a deterministic cutoff instead of depending on the host clock. + let expected_gc_root = &snapshots[1].0; + catalog + .set_table_lvt( + &LeastVisibleTimeIdent::new(ctx.get_tenant(), fuse_table.get_id()), + &LeastVisibleTime::new(expected_gc_root.timestamp.unwrap()), + ) + .await?; + + let table_ctx: Arc = ctx; + let selection = fuse_table + .prepare_snapshot_gc_selection(&table_ctx, true) + .await? + .expect("the persisted LVT should select a flashback GC root"); + + assert_eq!(selection.gc_root.snapshot_id, expected_gc_root.snapshot_id); + assert_eq!(selection.gc_root.timestamp, expected_gc_root.timestamp); + + Ok(()) +} + /// Regression test for chunked reads of protected gc-root segments. /// /// `do_vacuum2` reads the gc-root's protected segments in chunks of diff --git a/src/query/storages/fuse/src/operations/vacuum.rs b/src/query/storages/fuse/src/operations/vacuum.rs index da5042957df..b3abb5dc22a 100644 --- a/src/query/storages/fuse/src/operations/vacuum.rs +++ b/src/query/storages/fuse/src/operations/vacuum.rs @@ -84,6 +84,22 @@ use crate::io::TableMetaLocationGenerator; /// the above risks will not exist. pub const ASSUMPTION_MAX_TXN_DURATION: Duration = Duration::days(3); +fn retention_cutoff( + now: DateTime, + latest_snapshot_timestamp: DateTime, + retention_period: TimeDelta, +) -> DateTime { + std::cmp::min(now - retention_period, latest_snapshot_timestamp) +} + +fn flashback_gc_root_lvt(respect_flash_back: bool, lvt: DateTime) -> Option> { + respect_flash_back.then_some(lvt) +} + +fn is_flashback_gc_root(snapshot_timestamp: Option>, lvt: DateTime) -> bool { + snapshot_timestamp.is_some_and(|timestamp| timestamp <= lvt) +} + pub struct SnapshotGcSelection { pub gc_root: Arc, pub snapshots_to_gc: Vec, @@ -436,9 +452,7 @@ impl FuseTable { return Ok(None); }; - if respect_flash_back { - respect_flash_back_with_lvt = Some(lvt); - } + respect_flash_back_with_lvt = flashback_gc_root_lvt(respect_flash_back, lvt); ctx.set_status_info(&format!( "Set LVT for table {}, elapsed: {:?}, LVT: {:?}", @@ -555,7 +569,7 @@ impl FuseTable { let latest_location = self.snapshot_loc().unwrap(); let gc_root = self .find_location(ctx, latest_location, |snapshot| { - snapshot.timestamp.is_some_and(|ts| ts <= lvt) + is_flashback_gc_root(snapshot.timestamp, lvt) }) .await .ok(); @@ -677,7 +691,7 @@ impl FuseTable { let catalog = ctx.get_default_catalog()?; // safe to unwrap, as we have checked the version is v4 let latest_ts = latest_snapshot.timestamp.unwrap(); - let lvt_point_candidate = std::cmp::min(Utc::now() - retention_period, latest_ts); + let lvt_point_candidate = retention_cutoff(Utc::now(), latest_ts, retention_period); let lvt_point = catalog .set_table_lvt( @@ -725,3 +739,46 @@ pub fn slice_summary(s: &[T]) -> String { format!("{:?}", s) } } + +#[cfg(test)] +mod tests { + use chrono::TimeZone; + + use super::*; + + #[test] + fn test_retention_cutoff_is_bounded_by_latest_snapshot() { + let now = Utc.with_ymd_and_hms(2025, 1, 10, 12, 0, 0).unwrap(); + let retention_period = TimeDelta::days(2); + + let latest_after_cutoff = Utc.with_ymd_and_hms(2025, 1, 9, 12, 0, 0).unwrap(); + assert_eq!( + retention_cutoff(now, latest_after_cutoff, retention_period), + Utc.with_ymd_and_hms(2025, 1, 8, 12, 0, 0).unwrap() + ); + + let latest_before_cutoff = Utc.with_ymd_and_hms(2025, 1, 7, 12, 0, 0).unwrap(); + assert_eq!( + retention_cutoff(now, latest_before_cutoff, retention_period), + latest_before_cutoff + ); + } + + #[test] + fn test_flashback_gc_root_respects_flag_and_lvt_boundary() { + let lvt = Utc.with_ymd_and_hms(2025, 1, 8, 12, 0, 0).unwrap(); + + assert_eq!(flashback_gc_root_lvt(false, lvt), None); + assert_eq!(flashback_gc_root_lvt(true, lvt), Some(lvt)); + assert!(!is_flashback_gc_root(None, lvt)); + assert!(!is_flashback_gc_root( + Some(lvt + TimeDelta::microseconds(1)), + lvt + )); + assert!(is_flashback_gc_root(Some(lvt), lvt)); + assert!(is_flashback_gc_root( + Some(lvt - TimeDelta::microseconds(1)), + lvt + )); + } +} diff --git a/tests/suites/8_faked_time_prepare/00_prepare/00_0000_dummy_prepare.result b/tests/suites/8_faked_time_prepare/00_prepare/00_0000_dummy_prepare.result deleted file mode 100755 index e69de29bb2d..00000000000 diff --git a/tests/suites/8_faked_time_prepare/00_prepare/00_0000_dummy_prepare.sh b/tests/suites/8_faked_time_prepare/00_prepare/00_0000_dummy_prepare.sh deleted file mode 100755 index 9fc3e56d245..00000000000 --- a/tests/suites/8_faked_time_prepare/00_prepare/00_0000_dummy_prepare.sh +++ /dev/null @@ -1,10 +0,0 @@ -#!/usr/bin/env bash - -CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -. "$CURDIR"/../../../shell_env.sh - -echo "create or replace DATABASE test_faketime" | bendsql_connect_root - -echo "create table test_faketime.t(c timestamp)" | bendsql_connect_root - -echo "insert into table test_faketime.t values(now())" | bendsql_connect_root diff --git a/tests/suites/8_faked_time_prepare/00_prepare/00_0001_vacuum2_prepare.result b/tests/suites/8_faked_time_prepare/00_prepare/00_0001_vacuum2_prepare.result deleted file mode 100644 index c142d1c73b8..00000000000 --- a/tests/suites/8_faked_time_prepare/00_prepare/00_0001_vacuum2_prepare.result +++ /dev/null @@ -1,2 +0,0 @@ ->>>> create or replace table test_vacuum2(a int); ->>>> insert into test_vacuum2 values(1); diff --git a/tests/suites/8_faked_time_prepare/00_prepare/00_0001_vacuum2_prepare.sh b/tests/suites/8_faked_time_prepare/00_prepare/00_0001_vacuum2_prepare.sh deleted file mode 100755 index 1a59b81f672..00000000000 --- a/tests/suites/8_faked_time_prepare/00_prepare/00_0001_vacuum2_prepare.sh +++ /dev/null @@ -1,8 +0,0 @@ -#!/usr/bin/env bash - -CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -. "$CURDIR"/../../../shell_env.sh - -stmt "create or replace table test_vacuum2(a int);" - -stmt "insert into test_vacuum2 values(1);" \ No newline at end of file diff --git a/tests/suites/8_faked_time_prepare/00_prepare/00_0002_vacuum2_respect_time_travel.result b/tests/suites/8_faked_time_prepare/00_prepare/00_0002_vacuum2_respect_time_travel.result deleted file mode 100644 index 25838f156f1..00000000000 --- a/tests/suites/8_faked_time_prepare/00_prepare/00_0002_vacuum2_respect_time_travel.result +++ /dev/null @@ -1,2 +0,0 @@ ->>>> create or replace table test_vacuum2_respect_time_travel(a int); ->>>> insert into test_vacuum2_respect_time_travel values(1); diff --git a/tests/suites/8_faked_time_prepare/00_prepare/00_0002_vacuum2_respect_time_travel.sh b/tests/suites/8_faked_time_prepare/00_prepare/00_0002_vacuum2_respect_time_travel.sh deleted file mode 100755 index b2dd08fdf33..00000000000 --- a/tests/suites/8_faked_time_prepare/00_prepare/00_0002_vacuum2_respect_time_travel.sh +++ /dev/null @@ -1,8 +0,0 @@ -#!/usr/bin/env bash - -CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -. "$CURDIR"/../../../shell_env.sh - -stmt "create or replace table test_vacuum2_respect_time_travel(a int);" - -stmt "insert into test_vacuum2_respect_time_travel values(1);" \ No newline at end of file diff --git a/tests/suites/9_faked_time/00_dummy_cases/00_0001_dummy.result b/tests/suites/9_faked_time/00_dummy_cases/00_0001_dummy.result deleted file mode 100644 index d86bac9de59..00000000000 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0001_dummy.result +++ /dev/null @@ -1 +0,0 @@ -OK diff --git a/tests/suites/9_faked_time/00_dummy_cases/00_0001_dummy.sh b/tests/suites/9_faked_time/00_dummy_cases/00_0001_dummy.sh deleted file mode 100755 index bb628e811a0..00000000000 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0001_dummy.sh +++ /dev/null @@ -1,29 +0,0 @@ -#!/usr/bin/env bash - -CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -. "$CURDIR"/../../../shell_env.sh - - -############################################################# -# check that the timestamp we inserted during prepare phase # -# is at least 2 days ago # -############################################################# - -# format of `faked` is "2024-05-22 03:00:35.977589" -c=$(echo "select c from test_faketime.t" | bendsql_connect_root) -now=$(echo "select now()" | bendsql_connect_root) - -# manually "time diff" -faked=$(date -d "$c" +%s) -current=$(date -d "$now" +%s) - -time_diff=$((current- faked)) - -time_diff_days=$(python3 -c "print($time_diff / 86400)") - -# Check if time difference is greater than 2 days -if python3 -c "import sys; sys.exit(0 if $time_diff_days > 2 else 1)"; then - echo "OK" -else - echo "assertion failure, time_diff_days is [$time_diff_days]" -fi diff --git a/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.result b/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.result deleted file mode 100644 index fb72acf68ee..00000000000 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.result +++ /dev/null @@ -1,14 +0,0 @@ ->>>> insert into test_vacuum2 values(2); ->>>> set data_retention_time_in_days = 2;truncate table test_vacuum2; ->>>> insert into test_vacuum2 values(3); ->>>> select count(*) from fuse_snapshot('default','test_vacuum2') -4 -<<<< ->>>> set data_retention_time_in_days = 0;call system$fuse_vacuum2('default','test_vacuum2'); ->>>> select count(*) from fuse_snapshot('default','test_vacuum2') -1 -<<<< ->>>> select * from test_vacuum2; -3 -<<<< ->>>> set data_retention_time_in_days = 1; diff --git a/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.sh b/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.sh deleted file mode 100755 index 57c8e5d299f..00000000000 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.sh +++ /dev/null @@ -1,25 +0,0 @@ -#!/usr/bin/env bash - -CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -. "$CURDIR"/../../../shell_env.sh - -stmt "insert into test_vacuum2 values(2);" - -stmt "set data_retention_time_in_days = 2;truncate table test_vacuum2;" - -# gc root -stmt "insert into test_vacuum2 values(3);" - -# should have 4 snapshots -query "select count(*) from fuse_snapshot('default','test_vacuum2')" - -stmt "set data_retention_time_in_days = 0;call system\$fuse_vacuum2('default','test_vacuum2');" - -# only the current snapshot remains -query "select count(*) from fuse_snapshot('default','test_vacuum2')" - -# verify the data -query "select * from test_vacuum2;" - -# restore default value -stmt "set data_retention_time_in_days = 1;" \ No newline at end of file diff --git a/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.result b/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.result deleted file mode 100644 index 14aee110822..00000000000 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.result +++ /dev/null @@ -1,14 +0,0 @@ ->>>> insert into test_vacuum2_respect_time_travel values(2); ->>>> set data_retention_time_in_days = 2;truncate table test_vacuum2_respect_time_travel; ->>>> insert into test_vacuum2_respect_time_travel values(3); ->>>> select count(*) from fuse_snapshot('default','test_vacuum2_respect_time_travel') -4 -<<<< ->>>> set data_retention_time_in_days = 0;call system$fuse_vacuum2('default','test_vacuum2_respect_time_travel',true); ->>>> select count(*) from fuse_snapshot('default','test_vacuum2_respect_time_travel') -1 -<<<< ->>>> select * from test_vacuum2_respect_time_travel; -3 -<<<< ->>>> set data_retention_time_in_days = 1; \ No newline at end of file diff --git a/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.sh b/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.sh deleted file mode 100755 index 3167a87c2d8..00000000000 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.sh +++ /dev/null @@ -1,25 +0,0 @@ -#!/usr/bin/env bash - -CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) -. "$CURDIR"/../../../shell_env.sh - -stmt "insert into test_vacuum2_respect_time_travel values(2);" - -stmt "set data_retention_time_in_days = 2;truncate table test_vacuum2_respect_time_travel;" - -# gc root -stmt "insert into test_vacuum2_respect_time_travel values(3);" - -# should have 4 snapshots -query "select count(*) from fuse_snapshot('default','test_vacuum2_respect_time_travel')" - -stmt "set data_retention_time_in_days = 0;call system\$fuse_vacuum2('default','test_vacuum2_respect_time_travel',true);" - -# only the current snapshot remains -query "select count(*) from fuse_snapshot('default','test_vacuum2_respect_time_travel')" - -# verify the data -query "select * from test_vacuum2_respect_time_travel;" - -# restore default value -stmt "set data_retention_time_in_days = 1;" \ No newline at end of file From e2030983f7ddfa7547f01c013cbb36a165becec1 Mon Sep 17 00:00:00 2001 From: dantengsky Date: Wed, 2 Sep 2026 12:02:25 +0800 Subject: [PATCH 4/8] refactor(storage): clarify vacuum LVT predicate --- .../storages/fuse/src/operations/vacuum.rs | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/src/query/storages/fuse/src/operations/vacuum.rs b/src/query/storages/fuse/src/operations/vacuum.rs index b3abb5dc22a..fb4be59cf99 100644 --- a/src/query/storages/fuse/src/operations/vacuum.rs +++ b/src/query/storages/fuse/src/operations/vacuum.rs @@ -96,7 +96,10 @@ fn flashback_gc_root_lvt(respect_flash_back: bool, lvt: DateTime) -> Option respect_flash_back.then_some(lvt) } -fn is_flashback_gc_root(snapshot_timestamp: Option>, lvt: DateTime) -> bool { +fn is_snapshot_at_or_before_lvt( + snapshot_timestamp: Option>, + lvt: DateTime, +) -> bool { snapshot_timestamp.is_some_and(|timestamp| timestamp <= lvt) } @@ -569,7 +572,7 @@ impl FuseTable { let latest_location = self.snapshot_loc().unwrap(); let gc_root = self .find_location(ctx, latest_location, |snapshot| { - is_flashback_gc_root(snapshot.timestamp, lvt) + is_snapshot_at_or_before_lvt(snapshot.timestamp, lvt) }) .await .ok(); @@ -765,18 +768,18 @@ mod tests { } #[test] - fn test_flashback_gc_root_respects_flag_and_lvt_boundary() { + fn test_snapshot_at_or_before_lvt_boundary() { let lvt = Utc.with_ymd_and_hms(2025, 1, 8, 12, 0, 0).unwrap(); assert_eq!(flashback_gc_root_lvt(false, lvt), None); assert_eq!(flashback_gc_root_lvt(true, lvt), Some(lvt)); - assert!(!is_flashback_gc_root(None, lvt)); - assert!(!is_flashback_gc_root( + assert!(!is_snapshot_at_or_before_lvt(None, lvt)); + assert!(!is_snapshot_at_or_before_lvt( Some(lvt + TimeDelta::microseconds(1)), lvt )); - assert!(is_flashback_gc_root(Some(lvt), lvt)); - assert!(is_flashback_gc_root( + assert!(is_snapshot_at_or_before_lvt(Some(lvt), lvt)); + assert!(is_snapshot_at_or_before_lvt( Some(lvt - TimeDelta::microseconds(1)), lvt )); From 7830d7214f7bfcee179290291bf4aae5b63f9b14 Mon Sep 17 00:00:00 2001 From: dantengsky Date: Wed, 2 Sep 2026 12:19:19 +0800 Subject: [PATCH 5/8] test(storage): compare vacuum2 flashback modes --- .../it/storages/fuse/operations/vacuum2.rs | 27 ++++++++++++------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs b/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs index 12548268510..6301ac7a77d 100644 --- a/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs +++ b/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs @@ -156,7 +156,6 @@ async fn test_vacuum2_respect_flash_back_selects_lvt_snapshot() -> anyhow::Resul "insert into {db_name}.{tbl_name} values ({value})" )) .await?; - tokio::time::sleep(Duration::from_millis(2)).await; } let ctx = fixture.new_query_ctx().await?; @@ -177,24 +176,34 @@ async fn test_vacuum2_respect_flash_back_selects_lvt_snapshot() -> anyhow::Resul .await?; assert_eq!(snapshots.len(), 3); - // History is newest-first. Persist the middle snapshot as LVT so set_lvt() - // returns a deterministic cutoff instead of depending on the host clock. - let expected_gc_root = &snapshots[1].0; + // The history is S3 -> S2 -> S1, newest first. Set LVT to S2. + let lvt_snapshot = &snapshots[1].0; + let oldest_snapshot = &snapshots[2].0; catalog .set_table_lvt( &LeastVisibleTimeIdent::new(ctx.get_tenant(), fuse_table.get_id()), - &LeastVisibleTime::new(expected_gc_root.timestamp.unwrap()), + &LeastVisibleTime::new(lvt_snapshot.timestamp.unwrap()), ) .await?; let table_ctx: Arc = ctx; + + // Without flashback protection, S2 is only an anchor from object listing. + // Vacuum uses its committed predecessor S1 as the GC root. let selection = fuse_table - .prepare_snapshot_gc_selection(&table_ctx, true) + .prepare_snapshot_gc_selection(&table_ctx, false) .await? - .expect("the persisted LVT should select a flashback GC root"); + .expect("S1 should be selected as the GC root"); + assert_eq!(selection.gc_root.snapshot_id, oldest_snapshot.snapshot_id); - assert_eq!(selection.gc_root.snapshot_id, expected_gc_root.snapshot_id); - assert_eq!(selection.gc_root.timestamp, expected_gc_root.timestamp); + // With flashback protection, vacuum follows the committed snapshot chain + // and selects the first snapshot at or before LVT, which is S2. + let selection = fuse_table + .prepare_snapshot_gc_selection(&table_ctx, true) + .await? + .expect("S2 should be selected as the GC root"); + assert_eq!(selection.gc_root.snapshot_id, lvt_snapshot.snapshot_id); + assert_eq!(selection.gc_root.timestamp, lvt_snapshot.timestamp); Ok(()) } From 7771bee6f791ef8c609374ddbc03f926747d3d9a Mon Sep 17 00:00:00 2001 From: dantengsky Date: Wed, 2 Sep 2026 12:32:53 +0800 Subject: [PATCH 6/8] test(storage): cover vacuum2 after flashback --- .../it/storages/fuse/operations/vacuum2.rs | 33 ++++++++++++++----- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs b/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs index 6301ac7a77d..abe34e03b65 100644 --- a/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs +++ b/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs @@ -176,28 +176,45 @@ async fn test_vacuum2_respect_flash_back_selects_lvt_snapshot() -> anyhow::Resul .await?; assert_eq!(snapshots.len(), 3); - // The history is S3 -> S2 -> S1, newest first. Set LVT to S2. + // Start with S3 -> S2 -> S1, then flash back to S2. S3 remains in + // storage, but the current committed chain is S2 -> S1. + let abandoned_snapshot = &snapshots[0].0; let lvt_snapshot = &snapshots[1].0; let oldest_snapshot = &snapshots[2].0; + fixture + .execute_command(&format!( + "alter table {db_name}.{tbl_name} flashback to (snapshot => '{}')", + lvt_snapshot.snapshot_id.simple() + )) + .await?; + + let table_ctx: Arc = ctx; + let table = catalog + .get_table(&table_ctx.get_tenant(), &db_name, tbl_name) + .await?; + let fuse_table = FuseTable::try_from_table(table.as_ref())?; + let current_snapshot = fuse_table.read_table_snapshot().await?.unwrap(); + assert_eq!(current_snapshot.snapshot_id, lvt_snapshot.snapshot_id); + assert_ne!(current_snapshot.snapshot_id, abandoned_snapshot.snapshot_id); + + // Fix LVT at S2 so the test does not depend on the host clock. catalog .set_table_lvt( - &LeastVisibleTimeIdent::new(ctx.get_tenant(), fuse_table.get_id()), + &LeastVisibleTimeIdent::new(table_ctx.get_tenant(), fuse_table.get_id()), &LeastVisibleTime::new(lvt_snapshot.timestamp.unwrap()), ) .await?; - let table_ctx: Arc = ctx; - - // Without flashback protection, S2 is only an anchor from object listing. - // Vacuum uses its committed predecessor S1 as the GC root. + // Without flashback protection, S2 is an anchor from object listing. + // Vacuum uses its predecessor S1 as the GC root. let selection = fuse_table .prepare_snapshot_gc_selection(&table_ctx, false) .await? .expect("S1 should be selected as the GC root"); assert_eq!(selection.gc_root.snapshot_id, oldest_snapshot.snapshot_id); - // With flashback protection, vacuum follows the committed snapshot chain - // and selects the first snapshot at or before LVT, which is S2. + // With flashback protection, vacuum walks the current committed chain and + // selects the first snapshot at or before LVT, which is S2. let selection = fuse_table .prepare_snapshot_gc_selection(&table_ctx, true) .await? From 9361e51b2d2d708f162248f3fd17a71a71b90f5f Mon Sep 17 00:00:00 2001 From: dantengsky Date: Wed, 2 Sep 2026 12:41:12 +0800 Subject: [PATCH 7/8] test(storage): show flashback-safe vacuum root --- .../it/storages/fuse/operations/vacuum2.rs | 41 ++++++++++--------- 1 file changed, 22 insertions(+), 19 deletions(-) diff --git a/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs b/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs index abe34e03b65..8253b90a3d6 100644 --- a/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs +++ b/src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs @@ -150,7 +150,7 @@ async fn test_vacuum2_respect_flash_back_selects_lvt_snapshot() -> anyhow::Resul .execute_command(&format!("create table {db_name}.{tbl_name} (c int)")) .await?; - for value in 1..=3 { + for value in 1..=4 { fixture .execute_command(&format!( "insert into {db_name}.{tbl_name} values ({value})" @@ -174,17 +174,17 @@ async fn test_vacuum2_respect_flash_back_selects_lvt_snapshot() -> anyhow::Resul ) .try_collect() .await?; - assert_eq!(snapshots.len(), 3); + assert_eq!(snapshots.len(), 4); - // Start with S3 -> S2 -> S1, then flash back to S2. S3 remains in - // storage, but the current committed chain is S2 -> S1. - let abandoned_snapshot = &snapshots[0].0; - let lvt_snapshot = &snapshots[1].0; - let oldest_snapshot = &snapshots[2].0; + // Start with S4 -> S3 -> S2 -> S1, then flash back to S2. Object + // listing still sees S4 and S3, but the current chain is S2 -> S1. + let lvt_snapshot = &snapshots[0].0; + let abandoned_gc_root = &snapshots[1].0; + let flashback_snapshot = &snapshots[2].0; fixture .execute_command(&format!( "alter table {db_name}.{tbl_name} flashback to (snapshot => '{}')", - lvt_snapshot.snapshot_id.simple() + flashback_snapshot.snapshot_id.simple() )) .await?; @@ -194,10 +194,10 @@ async fn test_vacuum2_respect_flash_back_selects_lvt_snapshot() -> anyhow::Resul .await?; let fuse_table = FuseTable::try_from_table(table.as_ref())?; let current_snapshot = fuse_table.read_table_snapshot().await?.unwrap(); - assert_eq!(current_snapshot.snapshot_id, lvt_snapshot.snapshot_id); - assert_ne!(current_snapshot.snapshot_id, abandoned_snapshot.snapshot_id); + assert_eq!(current_snapshot.snapshot_id, flashback_snapshot.snapshot_id); - // Fix LVT at S2 so the test does not depend on the host clock. + // Fix LVT at S4. The persisted LVT is monotonic, so set_lvt() keeps this + // value even though the current snapshot is S2. catalog .set_table_lvt( &LeastVisibleTimeIdent::new(table_ctx.get_tenant(), fuse_table.get_id()), @@ -205,22 +205,25 @@ async fn test_vacuum2_respect_flash_back_selects_lvt_snapshot() -> anyhow::Resul ) .await?; - // Without flashback protection, S2 is an anchor from object listing. - // Vacuum uses its predecessor S1 as the GC root. + // Without flashback protection, object listing uses S4 as the anchor and + // selects its predecessor S3, which belongs to the abandoned branch. let selection = fuse_table .prepare_snapshot_gc_selection(&table_ctx, false) .await? - .expect("S1 should be selected as the GC root"); - assert_eq!(selection.gc_root.snapshot_id, oldest_snapshot.snapshot_id); + .expect("S3 should be selected as the GC root"); + assert_eq!(selection.gc_root.snapshot_id, abandoned_gc_root.snapshot_id); - // With flashback protection, vacuum walks the current committed chain and - // selects the first snapshot at or before LVT, which is S2. + // With flashback protection, vacuum walks the current committed chain from + // S2. It selects S2 and does not use a snapshot from the abandoned branch. let selection = fuse_table .prepare_snapshot_gc_selection(&table_ctx, true) .await? .expect("S2 should be selected as the GC root"); - assert_eq!(selection.gc_root.snapshot_id, lvt_snapshot.snapshot_id); - assert_eq!(selection.gc_root.timestamp, lvt_snapshot.timestamp); + assert_eq!( + selection.gc_root.snapshot_id, + flashback_snapshot.snapshot_id + ); + assert_eq!(selection.gc_root.timestamp, flashback_snapshot.timestamp); Ok(()) } From 9f40f25c1faebb7e41ad3e2a3ad3ebebe32f4d8d Mon Sep 17 00:00:00 2001 From: dantengsky Date: Wed, 2 Sep 2026 12:45:19 +0800 Subject: [PATCH 8/8] test(storage): explain retention cutoff cases --- src/query/storages/fuse/src/operations/vacuum.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/query/storages/fuse/src/operations/vacuum.rs b/src/query/storages/fuse/src/operations/vacuum.rs index fb4be59cf99..6886f694abc 100644 --- a/src/query/storages/fuse/src/operations/vacuum.rs +++ b/src/query/storages/fuse/src/operations/vacuum.rs @@ -754,12 +754,16 @@ mod tests { let now = Utc.with_ymd_and_hms(2025, 1, 10, 12, 0, 0).unwrap(); let retention_period = TimeDelta::days(2); + // When the latest snapshot is newer than the retention boundary, use + // now - retention as the cutoff. let latest_after_cutoff = Utc.with_ymd_and_hms(2025, 1, 9, 12, 0, 0).unwrap(); assert_eq!( retention_cutoff(now, latest_after_cutoff, retention_period), Utc.with_ymd_and_hms(2025, 1, 8, 12, 0, 0).unwrap() ); + // The cutoff must not move past the latest snapshot. This also keeps + // the result valid when the host clock is ahead of snapshot time. let latest_before_cutoff = Utc.with_ymd_and_hms(2025, 1, 7, 12, 0, 0).unwrap(); assert_eq!( retention_cutoff(now, latest_before_cutoff, retention_period),