update

Author: zhyass

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

@@ -33,15 +33,11 @@ use databend_common_meta_app::schema::ListHistoryTableBranchesReq;
 use databend_common_meta_app::schema::ListIndexesByIdReq;
 use databend_common_meta_app::schema::ListTableTagsReq;
 use databend_common_storages_fuse::FuseTable;
-use databend_common_storages_fuse::io::SegmentsIO;
 use databend_common_storages_fuse::io::SnapshotsIO;
 use databend_common_storages_fuse::io::TableMetaLocationGenerator;
 use databend_common_storages_fuse::operations::SnapshotGcSelection;
 use databend_meta_types::MatchSeq;
-use databend_storages_common_cache::CacheAccessor;
-use databend_storages_common_cache::CacheManager;
 use databend_storages_common_io::Files;
-use databend_storages_common_table_meta::meta::CompactSegmentInfo;
 use databend_storages_common_table_meta::meta::Location;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::meta::try_extract_uuid_str_from_path;
@@ -283,14 +279,14 @@ pub async fn do_vacuum2(
                 branch_table.as_ref(),
                 ctx.clone(),
                 None,
-                list_owner_files_to_gc(
-                    branch_table.as_ref(),
-                    branch_table
-                        .meta_location_generator()
-                        .snapshot_location_prefix(),
-                    None,
-                )
-                .await?,
+                branch_table
+                    .list_files_for_gc(
+                        branch_table
+                            .meta_location_generator()
+                            .snapshot_location_prefix(),
+                        None,
+                    )
+                    .await?,
                 protected_segments,
                 protected_blocks,
             )
@@ -434,17 +430,8 @@ async fn select_owner_gc_root(
         }
         None => {
             let gc_root_meta_ts = fuse_table
-                .get_operator_ref()
-                .stat(&latest_snapshot_loc)
-                .await
-                .map_err(ErrorCode::from)?
-                .last_modified()
-                .ok_or_else(|| {
-                    ErrorCode::StorageOther(format!(
-                        "Failed to get `last_modified` metadata of the snapshot object '{}'",
-                        latest_snapshot_loc
-                    ))
-                })?;
+                .snapshot_last_modified(&latest_snapshot_loc)
+                .await?;
             (
                 latest_snapshot,
                 latest_snapshot_loc.clone(),
@@ -465,26 +452,6 @@ async fn select_owner_gc_root(
     }))
 }
 
-async fn remove_snapshots(
-    fuse_table: &FuseTable,
-    ctx: Arc<dyn TableContext>,
-    snapshots_to_gc: &[String],
-) -> Result<()> {
-    if snapshots_to_gc.is_empty() {
-        return Ok(());
-    }
-
-    if let Some(snapshot_cache) = CacheManager::instance().get_table_snapshot_cache() {
-        for path in snapshots_to_gc {
-            snapshot_cache.evict(path);
-        }
-    }
-    Files::create(ctx, fuse_table.get_operator())
-        .remove_file_in_batch(snapshots_to_gc)
-        .await?;
-    Ok(())
-}
-
 fn merge_protected_segments(
     owner_prefixes: &OwnerPrefixes,
     owner_result: &OwnerSnapshotResult,
@@ -507,31 +474,23 @@ async fn collect_protected_blocks_by_owner(
     owner_prefixes: &OwnerPrefixes,
     protected_segments_by_owner: &ProtectedSegmentsByOwner,
 ) -> Result<ProtectedBlocksByOwner> {
-    let segments_io = SegmentsIO::create(
-        ctx.clone(),
-        base_fuse_table.get_operator(),
-        base_fuse_table.schema(),
-    );
     let mut protected_blocks_by_owner: ProtectedBlocksByOwner = HashMap::new();
 
     for protected_segments in protected_segments_by_owner.values() {
         if protected_segments.is_empty() {
             continue;
         }
 
-        let locations = protected_segments.iter().cloned().collect::<Vec<_>>();
-        let segments = segments_io
-            .read_segments::<Arc<CompactSegmentInfo>>(&locations, false)
+        let segment_refs = protected_segments.iter().collect::<Vec<_>>();
+        let block_locations = base_fuse_table
+            .get_block_locations(ctx.clone(), &segment_refs, false, false)
             .await?;
-        for segment in segments {
-            for block in segment?.block_metas()?.iter() {
-                let block_path = &block.location.0;
-                let owner_id = owner_id_by_path(owner_prefixes, block_path)?;
-                protected_blocks_by_owner
-                    .entry(owner_id)
-                    .or_default()
-                    .insert(try_extract_uuid_str_from_path(block_path)?.to_string());
-            }
+        for block_path in block_locations.block_location {
+            let owner_id = owner_id_by_path(owner_prefixes, &block_path)?;
+            protected_blocks_by_owner
+                .entry(owner_id)
+                .or_default()
+                .insert(try_extract_uuid_str_from_path(&block_path)?.to_string());
         }
     }
 
@@ -550,29 +509,29 @@ async fn cleanup_owner_data(
     let fuse_table = FuseTable::try_from_table(table)?;
     let table_info = fuse_table.get_table_info();
 
-    let segments_to_gc = list_owner_files_to_gc(
-        fuse_table,
-        fuse_table
-            .meta_location_generator()
-            .segment_location_prefix(),
-        cutoff,
-    )
-    .await?
-    .into_iter()
-    .filter(|path| {
-        !protected_segments.contains(&(
-            path.clone(),
-            TableMetaLocationGenerator::snapshot_version(path),
-        ))
-    })
-    .collect::<Vec<_>>();
-    let blocks_to_gc = filter_blocks_to_gc(
-        list_owner_files_to_gc(
-            fuse_table,
-            fuse_table.meta_location_generator().block_location_prefix(),
+    let segments_to_gc = fuse_table
+        .list_files_for_gc(
+            fuse_table
+                .meta_location_generator()
+                .segment_location_prefix(),
             cutoff,
         )
-        .await?,
+        .await?
+        .into_iter()
+        .filter(|path| {
+            !protected_segments.contains(&(
+                path.clone(),
+                TableMetaLocationGenerator::snapshot_version(path),
+            ))
+        })
+        .collect::<Vec<_>>();
+    let blocks_to_gc = filter_blocks_to_gc(
+        fuse_table
+            .list_files_for_gc(
+                fuse_table.meta_location_generator().block_location_prefix(),
+                cutoff,
+            )
+            .await?,
         &protected_blocks,
     )?;
 
@@ -636,7 +595,9 @@ async fn finalize_owner_snapshot_phase(
     protected_segments: HashSet<Location>,
     snapshot_files_to_gc: Vec<String>,
 ) -> Result<OwnerSnapshotResult> {
-    remove_snapshots(fuse_table, ctx, &snapshot_files_to_gc).await?;
+    fuse_table
+        .cleanup_snapshot_files(&ctx, &snapshot_files_to_gc, false)
+        .await?;
     let gc_root_snapshot_ts = gc_root.gc_root.timestamp.ok_or_else(|| {
         ErrorCode::IllegalReference(format!(
             "Table {} snapshot lacks required timestamp",
@@ -654,23 +615,6 @@ async fn finalize_owner_snapshot_phase(
     })
 }
 
-#[async_backtrace::framed]
-async fn list_owner_files_to_gc(
-    fuse_table: &FuseTable,
-    prefix: &str,
-    cutoff: Option<(DateTime<Utc>, DateTime<Utc>)>,
-) -> Result<Vec<String>> {
-    match cutoff {
-        Some((gc_root_timestamp, gc_root_meta_ts)) => Ok(fuse_table
-            .list_files_until_timestamp(prefix, gc_root_timestamp, false, Some(gc_root_meta_ts))
-            .await?
-            .into_iter()
-            .map(|entry| entry.path().to_string())
-            .collect()),
-        None => fuse_table.list_files(prefix.to_string(), |_, _| true).await,
-    }
-}
-
 #[async_backtrace::framed]
 async fn cleanup_owner_files(
     fuse_table: &FuseTable,
@@ -728,7 +672,9 @@ async fn cleanup_owner_files(
     let op = Files::create(ctx.clone(), fuse_table.get_operator());
     op.remove_file_in_batch(&indexes_to_gc).await?;
     op.remove_file_in_batch(&subject_files_to_gc).await?;
-    remove_snapshots(fuse_table, ctx, &snapshots_to_gc).await?;
+    fuse_table
+        .cleanup_snapshot_files(&ctx, &snapshots_to_gc, false)
+        .await?;
 
     Ok(subject_files_to_gc
         .into_iter()

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

@@ -976,17 +976,15 @@ impl Table for FuseTable {
         num_snapshot_limit: Option<usize>,
         dry_run: bool,
     ) -> Result<Option<Vec<String>>> {
-        match self.navigate_for_purge(&ctx, instant).await {
-            Ok((table, files)) => {
-                table
-                    .do_purge(&ctx, files, num_snapshot_limit, dry_run)
-                    .await
-            }
+        match self
+            .do_ref_aware_purge(&ctx, instant, num_snapshot_limit, dry_run)
+            .await
+        {
             Err(e) if e.code() == ErrorCode::TABLE_HISTORICAL_DATA_NOT_FOUND => {
                 warn!("navigate failed: {:?}", e);
                 if dry_run { Ok(Some(vec![])) } else { Ok(None) }
             }
-            Err(e) => Err(e),
+            res => res,
         }
     }
 

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

@@ -26,10 +26,12 @@ mod merge_into;
 mod mutation;
 mod mutation_source;
 mod navigate;
+mod purge;
 pub mod read;
 mod read_data;
 mod read_partitions;
 mod recluster;
+mod ref_gc;
 mod replace;
 mod replace_into;
 mod revert;