fix: purge table statistic files

Author: lichuang

src/query/service/tests/it/storages/fuse/operations/optimize.rs

@@ -48,6 +48,33 @@ async fn test_fuse_snapshot_optimize_statistic() -> Result<()> {
     .await
 }
 
+#[tokio::test]
+async fn test_fuse_snapshot_optimize_statistic_purge() -> Result<()> {
+    let fixture = TestFixture::new().await;
+    let db = fixture.default_db_name();
+    let tbl = fixture.default_table_name();
+    let case_name = "optimize_statistic_purge";
+    do_insertions(&fixture).await?;
+
+    // optimize statistics twice
+    for i in 0..1 {
+        let qry = format!("optimize table {}.{} statistic", db, tbl);
+
+        let ctx = fixture.ctx();
+        execute_command(ctx, &qry).await?;
+
+        check_data_dir(&fixture, case_name, 3, 1 + i, 2, 2, 2, Some(())).await?;
+    }
+
+    // After compact, all the count will become 1
+    let qry = format!("optimize table {}.{} all", db, tbl);
+    execute_command(fixture.ctx().clone(), &qry).await?;
+
+    check_data_dir(&fixture, case_name, 1, 1, 1, 1, 1, Some(())).await?;
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_fuse_snapshot_optimize_all() -> Result<()> {
     do_purge_test("explicit pure", "all", 1, 0, 1, 1, 1, None).await

src/query/service/tests/it/storages/fuse/table_test_fixture.rs

@@ -483,10 +483,12 @@ pub async fn check_data_dir(
     let prefix_block = FUSE_TBL_BLOCK_PREFIX;
     let prefix_index = FUSE_TBL_XOR_BLOOM_INDEX_PREFIX;
     let prefix_last_snapshot_hint = FUSE_TBL_LAST_SNAPSHOT_HINT;
+    println!("root: {}", root);
     for entry in WalkDir::new(root) {
         let entry = entry.unwrap();
         if entry.file_type().is_file() {
             let (_, entry_path) = entry.path().to_str().unwrap().split_at(root.len());
+            println!("entry: {}", entry_path);
             // trim the leading prefix, e.g. "/db_id/table_id/"
             let path = entry_path.split('/').skip(3).collect::<Vec<_>>();
             let path = path[0];
@@ -499,6 +501,7 @@ pub async fn check_data_dir(
             } else if path.starts_with(prefix_index) {
                 i_count += 1;
             } else if path.starts_with(prefix_snapshot_statistics) {
+                println!("ts file:{}", entry_path);
                 ts_count += 1;
             } else if path.starts_with(prefix_last_snapshot_hint) && check_last_snapshot.is_some() {
                 let content = fixture

src/query/storages/fuse/fuse/src/io/snapshots.rs

@@ -102,6 +102,21 @@ impl SnapshotsIO {
             .map_err(|e| ErrorCode::StorageOther(format!("read snapshots failure, {}", e)))
     }
 
+    // Read all the snapshots by the root file.
+    // limit: read how many snapshot files
+    // with_segment_locations: if true will get the segments of the snapshot
+    pub async fn read_table_statistic_files(
+        &self,
+        root_ts_file: &str,
+        limit: Option<usize>,
+    ) -> Result<Vec<String>> {
+        // Get all file list.
+        if let Some(prefix) = Self::get_s3_prefix_from_file(root_ts_file) {
+            return self.get_files(&prefix, limit, Some(root_ts_file)).await;
+        }
+        Ok(vec![])
+    }
+
     // Read all the snapshots by the root file.
     // limit: read how many snapshot files
     // with_segment_locations: if true will get the segments of the snapshot
@@ -123,7 +138,7 @@ impl SnapshotsIO {
         let mut snapshot_files = vec![];
         let mut segment_locations = HashSet::new();
         if let Some(prefix) = Self::get_s3_prefix_from_file(&root_snapshot_file) {
-            snapshot_files = self.get_files(&prefix, limit).await?;
+            snapshot_files = self.get_files(&prefix, limit, None).await?;
         }
 
         // 1. Get all the snapshot by chunks.
@@ -199,14 +214,22 @@ impl SnapshotsIO {
         Ok((snapshot_chain, segment_locations))
     }
 
-    async fn get_files(&self, prefix: &str, limit: Option<usize>) -> Result<Vec<String>> {
+    async fn get_files(
+        &self,
+        prefix: &str,
+        limit: Option<usize>,
+        exclude_file: Option<&str>,
+    ) -> Result<Vec<String>> {
         let data_accessor = self.operator.clone();
 
         let mut file_list = vec![];
         let mut ds = data_accessor.object(prefix).list().await?;
         while let Some(de) = ds.try_next().await? {
             match de.mode().await? {
                 ObjectMode::FILE => {
+                    if exclude_file.is_some() && Some(de.path()) == exclude_file {
+                        continue;
+                    }
                     let location = de.path().to_string();
                     let modified = de.last_modified().await?;
                     file_list.push((location, modified));

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

@@ -58,14 +58,20 @@ impl FuseTable {
         // 1. Root snapshot.
         let mut segments_referenced_by_root = HashSet::new();
         let mut locations_referenced_by_root = Default::default();
-        let (root_snapshot_id, root_snapshot_ts) = if let Some(root_snapshot) = snapshot_opt {
-            let segments = root_snapshot.segments.clone();
-            locations_referenced_by_root = self.get_block_locations(ctx.clone(), &segments).await?;
-            segments_referenced_by_root = HashSet::from_iter(segments);
-            (root_snapshot.snapshot_id, root_snapshot.timestamp)
-        } else {
-            (SnapshotId::new_v4(), None)
-        };
+        let (root_snapshot_id, root_snapshot_ts, root_ts_location_opt) =
+            if let Some(ref root_snapshot) = snapshot_opt {
+                let segments = root_snapshot.segments.clone();
+                locations_referenced_by_root =
+                    self.get_block_locations(ctx.clone(), &segments).await?;
+                segments_referenced_by_root = HashSet::from_iter(segments);
+                (
+                    root_snapshot.snapshot_id,
+                    root_snapshot.timestamp,
+                    root_snapshot.table_statistics_location.clone(),
+                )
+            } else {
+                (SnapshotId::new_v4(), None, None)
+            };
 
         // 2. Get all snapshot(including root snapshot).
         let mut all_snapshot_lites = vec![];
@@ -108,6 +114,7 @@ impl FuseTable {
         // 3. Find.
         let mut snapshots_to_be_purged = HashSet::new();
         let mut segments_to_be_purged = HashSet::new();
+        let ts_to_be_purged: Vec<String> = vec![];
 
         // 3.1 Find all the snapshots need to be deleted.
         {
@@ -131,6 +138,29 @@ impl FuseTable {
             }
         }
 
+        // 3.3 Find all the table statistic files need to be deleted
+        {
+            if let Some(root_ts_location) = root_ts_location_opt {
+                let start = Instant::now();
+                let snapshots_io = SnapshotsIO::create(
+                    ctx.clone(),
+                    self.operator.clone(),
+                    self.snapshot_format_version().await?,
+                );
+                let ts_to_be_purged = snapshots_io
+                    .read_table_statistic_files(&root_ts_location, None)
+                    .await?;
+                let status_ts_scan_count = ts_to_be_purged.len();
+                let status_ts_scan_cost = start.elapsed().as_secs();
+                let status = format!(
+                    "gc: scan table statistic files:{} takes:{} sec.",
+                    status_ts_scan_count, status_ts_scan_cost,
+                );
+                self.data_metrics.set_status(&status);
+                info!(status);
+            }
+        }
+
         let chunk_size = ctx.get_settings().get_max_storage_io_requests()? as usize;
 
         // 4. Purge segments&blocks by chunk size
@@ -240,6 +270,33 @@ impl FuseTable {
             }
         }
 
+        // 6. Purge table statistic files
+        {
+            let mut status_purged_count = 0;
+            let status_need_purged_count = ts_to_be_purged.len();
+            let start = Instant::now();
+            for chunk in ts_to_be_purged.chunks(chunk_size) {
+                let mut ts_locations_to_be_purged = HashSet::new();
+                for file in chunk {
+                    ts_locations_to_be_purged.insert(file.clone());
+                }
+                self.try_purge_location_files(ctx.clone(), ts_locations_to_be_purged)
+                    .await?;
+                // Refresh status.
+                {
+                    status_purged_count += chunk.len();
+                    let status = format!(
+                        "gc: table statistic files need to be purged:{}, have purged:{}, take:{} sec",
+                        status_need_purged_count,
+                        status_purged_count,
+                        start.elapsed().as_secs()
+                    );
+                    self.data_metrics.set_status(&status);
+                    info!(status);
+                }
+            }
+        }
+
         Ok(())
     }
 

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

@@ -92,6 +92,7 @@ impl FuseTable {
                     table_statistics.format_version(),
                 )?;
 
+            println!("ts location: {:?}", table_statistics);
             // 4. Save table statistics
             let mut new_snapshot = TableSnapshot::from_previous(&snapshot);
             new_snapshot.table_statistics_location = Some(table_statistics_location);