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/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..8253b90a3d6 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,99 @@ 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..=4 { + fixture + .execute_command(&format!( + "insert into {db_name}.{tbl_name} values ({value})" + )) + .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(), 4); + + // 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 => '{}')", + flashback_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, flashback_snapshot.snapshot_id); + + // 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()), + &LeastVisibleTime::new(lvt_snapshot.timestamp.unwrap()), + ) + .await?; + + // 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("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 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, + flashback_snapshot.snapshot_id + ); + assert_eq!(selection.gc_root.timestamp, flashback_snapshot.timestamp); + + Ok(()) +} + /// Regression test for chunked reads of protected gc-root segments. /// /// `do_vacuum2` reads the gc-root's protected segments in chunks of @@ -220,12 +319,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/src/query/storages/fuse/src/operations/vacuum.rs b/src/query/storages/fuse/src/operations/vacuum.rs index da5042957df..6886f694abc 100644 --- a/src/query/storages/fuse/src/operations/vacuum.rs +++ b/src/query/storages/fuse/src/operations/vacuum.rs @@ -84,6 +84,25 @@ 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_snapshot_at_or_before_lvt( + 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 +455,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 +572,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_snapshot_at_or_before_lvt(snapshot.timestamp, lvt) }) .await .ok(); @@ -677,7 +694,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 +742,50 @@ 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); + + // 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), + latest_before_cutoff + ); + } + + #[test] + 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_snapshot_at_or_before_lvt(None, lvt)); + assert!(!is_snapshot_at_or_before_lvt( + Some(lvt + TimeDelta::microseconds(1)), + lvt + )); + assert!(is_snapshot_at_or_before_lvt(Some(lvt), lvt)); + assert!(is_snapshot_at_or_before_lvt( + Some(lvt - TimeDelta::microseconds(1)), + lvt + )); + } +} 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 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 5f5a2cf6a2f..00000000000 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.result +++ /dev/null @@ -1,13 +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 -<<<< ->>>> 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 7bae89aa08f..00000000000 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0002_vacuum2.sh +++ /dev/null @@ -1,53 +0,0 @@ -#!/usr/bin/env bash - -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 - -# remain two snapshots -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 0163305ee37..00000000000 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.result +++ /dev/null @@ -1,13 +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 -<<<< ->>>> 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 3e4269ade91..00000000000 --- a/tests/suites/9_faked_time/00_dummy_cases/00_0003_vacuum2_respect_time_travel.sh +++ /dev/null @@ -1,53 +0,0 @@ -#!/usr/bin/env bash - -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 - -# remain two snapshots -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