Merge pull request #9119 from lichuang/purge_ts_files

fix: purge table statistic files

Author: BohuTANG

docs/doc/14-sql-commands/00-ddl/20-table/60-optimize-table.md

@@ -5,10 +5,10 @@ title: OPTIMIZE TABLE
 The objective of optimizing a table in Databend is to compact or purge its historical data in your object storage. This helps save storage space and improve query efficiency.
 
 :::caution
-Databend's Time Travel feature relies on historical data. If you purge historical data from a table with the command `OPTIMIZE TABLE <your_table> PURGE` or `OPTIMIZE TABLE <your_table> ALL`, the table will not be eligible for time travel. The command removes all snapshots (except the most recent one) and their associated segments and block files.
+Databend's Time Travel feature relies on historical data. If you purge historical data from a table with the command `OPTIMIZE TABLE <your_table> PURGE` or `OPTIMIZE TABLE <your_table> ALL`, the table will not be eligible for time travel. The command removes all snapshots (except the most recent one) and their associated segments,block files and table statistic file.
 :::
 
-## What are Snapshot, Segment, and Block?
+## What are Snapshot, Segment, Block and Table statistic file?
 
 Snapshot, segment, and block are the concepts Databend uses for data storage. Databend uses them to construct a hierarchical structure for storing table data.
 
@@ -20,6 +20,8 @@ A snapshot is a JSON file that does not save the table's data but indicate the s
 
 A segment is a JSON file that organizes the storage blocks (at least 1, at most 1,000) where the data is stored. If you run [FUSE_SEGMENT](../../../15-sql-functions/111-system-functions/fuse_segment.md) against a snapshot with the snapshot ID, you can find which segments are referenced by the snapshot.
 
+A table statistic file is a JSON file that save table statistic data, such as distinct values of table column.
+
 Databends saves actual table data in parquet files and considers each parquet file as a block. If you run [FUSE_BLOCK](../../../15-sql-functions/111-system-functions/fuse_block.md) against a snapshot with the snapshot ID, you can find which blocks are referenced by the snapshot.
 
 Databend creates a unique ID for each database and table for storing the snapshot, segment, and block files and saves them to your object storage in the path `<bucket_name>/[root]/<db_id>/<table_id>/`. Each snapshot, segment, and block file is named with a UUID (32-character lowercase hexadecimal string).
@@ -29,6 +31,7 @@ Databend creates a unique ID for each database and table for storing the snapsho
 | Snapshot | JSON    | `<32bitUUID>_<version>.json`    | `<bucket_name>/[root]/<db_id>/<table_id>/_ss/`   |
 | Segment  | JSON    | `<32bitUUID>_<version>.json`    | `<bucket_name>/[root]/<db_id>/<table_id>/_sg/`   |
 | Block    | parquet | `<32bitUUID>_<version>.parquet` | `<bucket_name>/[root]/<db_id>/<table_id>/_b/` |
+| Table statistic | JSON    | `<32bitUUID>_<version>.json`    | `<bucket_name>/[root]/<db_id>/<table_id>/_ts/`   |
 
 ## Table Optimization Considerations
 
@@ -69,7 +72,7 @@ OPTIMIZE TABLE [database.]table_name [ PURGE | COMPACT | ALL | STATISTIC ] [SEGM
 
 - `OPTIMIZE TABLE <table_name> PURGE`
 
-    Purges the historical data of table. Only the latest snapshot (including the segments and blocks referenced by this snapshot) will be kept.
+    Purges the historical data of table. Only the latest snapshot (including the segments, blocks and table statistic file referenced by this snapshot) will be kept.
 
 - `OPTIMIZE TABLE <table_name> COMPACT [LIMIT <segment_count>]`
 

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(()), None).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(()), 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
@@ -84,6 +111,7 @@ async fn do_purge_test(
         block_count,
         index_count,
         Some(()),
+        None,
     )
     .await?;
     history_should_have_item(&fixture, case_name, snapshot_count).await?;
@@ -103,6 +131,7 @@ async fn do_purge_test(
             block_count,
             index_count,
             Some(()),
+            None,
         )
         .await?;
 

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

@@ -59,6 +59,7 @@ async fn test_fuse_snapshot_truncate_in_drop_all_stmt() -> Result<()> {
         0, // 0 blocks
         0, // 0 index
         None,
+        None,
     )
     .await?;
     Ok(())

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

@@ -43,6 +43,7 @@ async fn test_fuse_truncate_purge_stmt() -> Result<()> {
         2,
         expected_index_count,
         Some(()),
+        None,
     )
     .await?;
 
@@ -68,6 +69,7 @@ async fn test_fuse_truncate_purge_stmt() -> Result<()> {
         0,
         0,
         Some(()),
+        None,
     )
     .await?;
     Ok(())

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

@@ -466,6 +466,7 @@ pub async fn check_data_dir(
     block_count: u32,
     index_count: u32,
     check_last_snapshot: Option<()>,
+    check_table_statistic_file: Option<()>,
 ) -> Result<()> {
     let data_path = match &GlobalConfig::instance().storage.params {
         StorageParams::Fs(v) => v.root.clone(),
@@ -478,6 +479,7 @@ pub async fn check_data_dir(
     let mut b_count = 0;
     let mut i_count = 0;
     let mut last_snapshot_loc = "".to_string();
+    let mut table_statistic_files = vec![];
     let prefix_snapshot = FUSE_TBL_SNAPSHOT_PREFIX;
     let prefix_snapshot_statistics = FUSE_TBL_SNAPSHOT_STATISTICS_PREFIX;
     let prefix_segment = FUSE_TBL_SEGMENT_PREFIX;
@@ -501,6 +503,7 @@ pub async fn check_data_dir(
                 i_count += 1;
             } else if path.starts_with(prefix_snapshot_statistics) {
                 ts_count += 1;
+                table_statistic_files.push(entry_path.to_string());
             } else if path.starts_with(prefix_last_snapshot_hint) && check_last_snapshot.is_some() {
                 let content = fixture
                     .ctx
@@ -553,6 +556,26 @@ pub async fn check_data_dir(
         );
     }
 
+    if check_table_statistic_file.is_some() {
+        let table = fixture.latest_default_table().await?;
+        let fuse_table = FuseTable::try_from_table(table.as_ref())?;
+        let snapshot_opt = fuse_table.read_table_snapshot().await?;
+        assert!(snapshot_opt.is_some());
+        let snapshot = snapshot_opt.unwrap();
+        let ts_location_opt = snapshot.table_statistics_location.clone();
+        assert!(ts_location_opt.is_some());
+        let ts_location = ts_location_opt.unwrap();
+        println!(
+            "ts_location_opt: {:?}, table_statistic_files: {:?}",
+            ts_location, table_statistic_files
+        );
+        assert!(
+            table_statistic_files
+                .iter()
+                .any(|e| e.contains(&ts_location))
+        );
+    }
+
     Ok(())
 }
 

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

@@ -102,6 +102,20 @@ impl SnapshotsIO {
             .map_err(|e| ErrorCode::StorageOther(format!("read snapshots failure, {}", e)))
     }
 
+    // Read all the table statistic files by the root file(exclude the root file).
+    // limit: read how many table statistic files
+    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 +137,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 +213,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(())
     }