feat: add table function fuse_vacuum2()

src/query/catalog/src/catalog/interface.rs

@@ -21,6 +21,7 @@ use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_meta_app::schema::database_name_ident::DatabaseNameIdent;
 use databend_common_meta_app::schema::dictionary_name_ident::DictionaryNameIdent;
+use databend_common_meta_app::schema::least_visible_time_ident::LeastVisibleTimeIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CommitTableMetaReply;
 use databend_common_meta_app::schema::CommitTableMetaReq;
@@ -63,6 +64,7 @@ use databend_common_meta_app::schema::GetSequenceReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::IndexMeta;
+use databend_common_meta_app::schema::LeastVisibleTime;
 use databend_common_meta_app::schema::ListDictionaryReq;
 use databend_common_meta_app::schema::ListDroppedTableReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
@@ -523,5 +525,13 @@ pub trait Catalog: DynClone + Send + Sync + Debug {
         req: ListDictionaryReq,
     ) -> Result<Vec<(String, DictionaryMeta)>>;
 
+    async fn set_table_lvt(
+        &self,
+        _name_ident: &LeastVisibleTimeIdent,
+        _value: &LeastVisibleTime,
+    ) -> Result<LeastVisibleTime> {
+        unimplemented!()
+    }
+
     async fn rename_dictionary(&self, req: RenameDictionaryReq) -> Result<()>;
 }

src/query/catalog/src/table.rs

@@ -38,6 +38,7 @@ use databend_common_pipeline_core::Pipeline;
 use databend_common_storage::Histogram;
 use databend_common_storage::StorageMetrics;
 use databend_storages_common_table_meta::meta::SnapshotId;
+use databend_storages_common_table_meta::meta::TableMetaTimestamps;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::table::ChangeType;
 use databend_storages_common_table_meta::table::OPT_KEY_TEMP_PREFIX;
@@ -234,7 +235,12 @@ pub trait Table: Sync + Send {
     }
 
     /// Assembly the pipeline of appending data to storage
-    fn append_data(&self, ctx: Arc<dyn TableContext>, pipeline: &mut Pipeline) -> Result<()> {
+    fn append_data(
+        &self,
+        ctx: Arc<dyn TableContext>,
+        pipeline: &mut Pipeline,
+        _table_meta_timestamps: TableMetaTimestamps,
+    ) -> Result<()> {
         let (_, _) = (ctx, pipeline);
 
         Err(ErrorCode::Unimplemented(format!(
@@ -253,6 +259,7 @@ pub trait Table: Sync + Send {
         overwrite: bool,
         prev_snapshot_id: Option<SnapshotId>,
         _deduplicated_label: Option<String>,
+        _table_meta_timestamps: TableMetaTimestamps,
     ) -> Result<()> {
         let (_, _, _, _, _, _) = (
             ctx,

src/query/catalog/src/table_context.rs

@@ -59,6 +59,8 @@ use databend_common_users::GrantObjectVisibilityChecker;
 use databend_storages_common_session::SessionState;
 use databend_storages_common_session::TxnManagerRef;
 use databend_storages_common_table_meta::meta::Location;
+use databend_storages_common_table_meta::meta::TableMetaTimestamps;
+use databend_storages_common_table_meta::meta::TableSnapshot;
 use parking_lot::Mutex;
 use parking_lot::RwLock;
 use xorf::BinaryFuse16;
@@ -328,6 +330,11 @@ pub trait TableContext: Send + Sync {
 
     fn has_bloom_runtime_filters(&self, id: usize) -> bool;
     fn txn_mgr(&self) -> TxnManagerRef;
+    fn get_table_meta_timestamps(
+        &self,
+        table_id: u64,
+        previous_snapshot: Option<Arc<TableSnapshot>>,
+    ) -> Result<TableMetaTimestamps>;
 
     fn get_read_block_thresholds(&self) -> BlockThresholds;
     fn set_read_block_thresholds(&self, _thresholds: BlockThresholds);

src/query/ee/Cargo.toml

@@ -65,6 +65,7 @@ jwt-simple = { workspace = true }
 log = { workspace = true }
 opendal = { workspace = true }
 tempfile = { workspace = true }
+uuid = { workspace = true }
 
 # aws sdk
 aws-config = { workspace = true, features = ["behavior-version-latest"] }

src/query/ee/src/storages/fuse/mod.rs

@@ -18,4 +18,5 @@ pub mod operations;
 pub use io::snapshots::get_snapshot_referenced_segments;
 pub use operations::vacuum_drop_tables::vacuum_drop_tables;
 pub use operations::vacuum_table::do_vacuum;
+pub use operations::vacuum_table_v2::do_vacuum2;
 pub use operations::virtual_columns::do_refresh_virtual_column;

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

@@ -28,6 +28,7 @@ use databend_enterprise_vacuum_handler::VacuumHandler;
 use databend_enterprise_vacuum_handler::VacuumHandlerWrapper;
 
 use crate::storages::fuse::do_vacuum;
+use crate::storages::fuse::operations::vacuum_table_v2::do_vacuum2;
 use crate::storages::fuse::operations::vacuum_temporary_files::do_vacuum_temporary_files;
 use crate::storages::fuse::vacuum_drop_tables;
 pub struct RealVacuumHandler {}
@@ -44,6 +45,15 @@ impl VacuumHandler for RealVacuumHandler {
         do_vacuum(fuse_table, ctx, retention_time, dry_run).await
     }
 
+    async fn do_vacuum2(
+        &self,
+        fuse_table: &FuseTable,
+        ctx: Arc<dyn TableContext>,
+        respect_flash_back: bool,
+    ) -> Result<Vec<String>> {
+        do_vacuum2(fuse_table, ctx, respect_flash_back).await
+    }
+
     async fn do_vacuum_drop_tables(
         &self,
         threads_nums: usize,

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

@@ -15,7 +15,7 @@
 pub mod handler;
 pub mod vacuum_drop_tables;
 pub mod vacuum_table;
+pub mod vacuum_table_v2;
 pub mod vacuum_temporary_files;
 pub mod virtual_columns;
-
 pub use handler::RealVacuumHandler;

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

@@ -177,7 +177,7 @@ pub async fn do_gc_orphan_files(
     let location_gen = fuse_table.meta_location_generator();
     let segment_locations_to_be_purged = get_orphan_files_to_be_purged(
         fuse_table,
-        location_gen.segment_info_prefix(),
+        location_gen.segment_location_prefix(),
         referenced_files.segments,
         retention_time,
     )
@@ -298,7 +298,7 @@ pub async fn do_dry_run_orphan_files(
     // 2. Get purge orphan segment files.
     let segment_locations_to_be_purged = get_orphan_files_to_be_purged(
         fuse_table,
-        location_gen.segment_info_prefix(),
+        location_gen.segment_location_prefix(),
         referenced_files.segments,
         retention_time,
     )