Merge pull request #9060 from lichuang/last_snapshot_bug

fix: update last_snapshot_hint file when purge

Author: mergify[bot]

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

@@ -12,6 +12,7 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+#![allow(clippy::too_many_arguments)]
 mod io;
 mod meta;
 mod misc;

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

@@ -83,8 +83,9 @@ async fn do_purge_test(
         segment_count,
         block_count,
         index_count,
+        Some(()),
     )
-    .await;
+    .await?;
     history_should_have_item(&fixture, case_name, snapshot_count).await?;
 
     if let Some((snapshot_count, table_statistic_count, segment_count, block_count, index_count)) =
@@ -101,8 +102,9 @@ async fn do_purge_test(
             segment_count,
             block_count,
             index_count,
+            Some(()),
         )
-        .await;
+        .await?;
 
         history_should_have_item(&fixture, case_name, snapshot_count).await?;
     };

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

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

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

@@ -34,7 +34,17 @@ async fn test_fuse_truncate_purge_stmt() -> Result<()> {
 
     let expected_index_count = 2;
     // there should be some data there: 2 snapshot, 2 segment, 2 block
-    check_data_dir(&fixture, "truncate_purge", 2, 0, 2, 2, expected_index_count).await;
+    check_data_dir(
+        &fixture,
+        "truncate_purge",
+        2,
+        0,
+        2,
+        2,
+        expected_index_count,
+        Some(()),
+    )
+    .await?;
 
     // let's truncate
     let qry = format!("truncate table {}.{} purge", db, tbl);
@@ -57,7 +67,8 @@ async fn test_fuse_truncate_purge_stmt() -> Result<()> {
         0,
         0,
         0,
+        Some(()),
     )
-    .await;
+    .await?;
     Ok(())
 }

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

@@ -13,6 +13,7 @@
 //  limitations under the License.
 
 use std::collections::VecDeque;
+use std::str;
 use std::sync::Arc;
 
 use common_ast::ast::Engine;
@@ -30,6 +31,7 @@ use common_sql::plans::CreateDatabasePlan;
 use common_sql::plans::CreateTablePlanV2;
 use common_storage::StorageFsConfig;
 use common_storage::StorageParams;
+use common_storages_fuse::FuseTable;
 use common_storages_fuse::FUSE_TBL_XOR_BLOOM_INDEX_PREFIX;
 use common_storages_table_meta::table::OPT_KEY_DATABASE_ID;
 use databend_query::interpreters::append2table;
@@ -46,6 +48,7 @@ use databend_query::sql::Planner;
 use databend_query::storages::fuse::table_functions::ClusteringInformationTable;
 use databend_query::storages::fuse::table_functions::FuseSnapshotTable;
 use databend_query::storages::fuse::FUSE_TBL_BLOCK_PREFIX;
+use databend_query::storages::fuse::FUSE_TBL_LAST_SNAPSHOT_HINT;
 use databend_query::storages::fuse::FUSE_TBL_SEGMENT_PREFIX;
 use databend_query::storages::fuse::FUSE_TBL_SNAPSHOT_PREFIX;
 use databend_query::storages::fuse::FUSE_TBL_SNAPSHOT_STATISTICS_PREFIX;
@@ -461,7 +464,8 @@ pub async fn check_data_dir(
     segment_count: u32,
     block_count: u32,
     index_count: u32,
-) {
+    check_last_snapshot: Option<()>,
+) -> Result<()> {
     let data_path = match fixture.ctx().get_config().storage.params {
         StorageParams::Fs(v) => v.root,
         _ => panic!("storage type is not fs"),
@@ -472,11 +476,13 @@ pub async fn check_data_dir(
     let mut sg_count = 0;
     let mut b_count = 0;
     let mut i_count = 0;
+    let mut last_snapshot_loc = "".to_string();
     let prefix_snapshot = FUSE_TBL_SNAPSHOT_PREFIX;
     let prefix_snapshot_statistics = FUSE_TBL_SNAPSHOT_STATISTICS_PREFIX;
     let prefix_segment = FUSE_TBL_SEGMENT_PREFIX;
     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;
     for entry in WalkDir::new(root) {
         let entry = entry.unwrap();
         if entry.file_type().is_file() {
@@ -494,6 +500,14 @@ pub async fn check_data_dir(
                 i_count += 1;
             } else if path.starts_with(prefix_snapshot_statistics) {
                 ts_count += 1;
+            } else if path.starts_with(prefix_last_snapshot_hint) && check_last_snapshot.is_some() {
+                let content = fixture
+                    .ctx
+                    .get_data_operator()?
+                    .object(entry_path)
+                    .read()
+                    .await?;
+                last_snapshot_loc = str::from_utf8(&content)?.to_string();
             }
         }
     }
@@ -525,6 +539,20 @@ pub async fn check_data_dir(
         "case [{}], check index count",
         case_name
     );
+
+    if check_last_snapshot.is_some() {
+        let table = fixture.latest_default_table().await?;
+        let fuse_table = FuseTable::try_from_table(table.as_ref())?;
+        let snapshot_loc = fuse_table.snapshot_loc().await?;
+        let snapshot_loc = snapshot_loc.unwrap();
+        assert!(last_snapshot_loc.contains(&snapshot_loc));
+        assert_eq!(
+            last_snapshot_loc.find(&snapshot_loc),
+            Some(last_snapshot_loc.len() - snapshot_loc.len())
+        );
+    }
+
+    Ok(())
 }
 
 pub async fn history_should_have_item(

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

@@ -443,7 +443,7 @@ impl FuseTable {
     }
 
     // Left a hint file which indicates the location of the latest snapshot
-    async fn write_last_snapshot_hint(
+    pub async fn write_last_snapshot_hint(
         operator: &Operator,
         location_generator: &TableMetaLocationGenerator,
         last_snapshot_path: String,

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

@@ -82,11 +82,25 @@ impl FuseTable {
 
             let start = Instant::now();
             (all_snapshot_lites, all_segment_locations) = snapshots_io
-                .read_snapshot_lites(root_snapshot_location, None, true, root_snapshot_ts, |x| {
-                    self.data_metrics.set_status(&x);
-                })
+                .read_snapshot_lites(
+                    root_snapshot_location.clone(),
+                    None,
+                    true,
+                    root_snapshot_ts,
+                    |x| {
+                        self.data_metrics.set_status(&x);
+                    },
+                )
                 .await?;
 
+            // try keep a hit file of last snapshot
+            Self::write_last_snapshot_hint(
+                &self.operator,
+                &self.meta_location_generator,
+                root_snapshot_location,
+            )
+            .await;
+
             status_snapshot_scan_count += all_snapshot_lites.len();
             status_snapshot_scan_cost += start.elapsed().as_secs();
         }

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

@@ -56,9 +56,10 @@ impl FuseTable {
 
             let mut new_table_meta = self.table_info.meta.clone();
             // update snapshot location
-            new_table_meta
-                .options
-                .insert(OPT_KEY_SNAPSHOT_LOCATION.to_owned(), new_snapshot_loc);
+            new_table_meta.options.insert(
+                OPT_KEY_SNAPSHOT_LOCATION.to_owned(),
+                new_snapshot_loc.clone(),
+            );
 
             // update table statistics, all zeros
             new_table_meta.statistics = TableStatistics::default();
@@ -80,6 +81,14 @@ impl FuseTable {
             catalog
                 .truncate_table(&tenant, &db_name, TruncateTableReq { table_id })
                 .await?;
+
+            // try keep a hit file of last snapshot
+            Self::write_last_snapshot_hint(
+                &self.operator,
+                &self.meta_location_generator,
+                new_snapshot_loc,
+            )
+            .await;
         }
 
         Ok(())