fix: purge may not work on tables after a flash back operation (#16812)

* fix: purge may not work on tables after a flash back operation

* fix

* fix test

* fix test

Author: SkyFan2002

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

@@ -168,7 +168,7 @@ async fn test_navigate_for_purge() -> Result<()> {
 
     // keep the first snapshot of the insertion
     let table = fixture.latest_default_table().await?;
-    let first_snapshot = FuseTable::try_from_table(table.as_ref())?
+    let _first_snapshot = FuseTable::try_from_table(table.as_ref())?
         .snapshot_loc()
         .await?
         .unwrap();
@@ -233,7 +233,7 @@ async fn test_navigate_for_purge() -> Result<()> {
     // navigate from the instant that is just one ms before the timestamp of the latest snapshot.
     let (navigate, files) = fuse_table.list_by_time_point(time_point).await?;
     assert_eq!(2, files.len());
-    assert_eq!(navigate, first_snapshot);
+    assert_eq!(navigate, third_snapshot);
 
     // 5. navigate by snapshot id.
     let snapshot_id = snapshots[1].0.snapshot_id.simple().to_string();

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

@@ -24,7 +24,6 @@ use databend_common_exception::Result;
 use databend_common_exception::ResultExt;
 use databend_common_meta_app::schema::TableInfo;
 use databend_common_meta_app::schema::TableStatistics;
-use databend_storages_common_cache::LoadParams;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::table::OPT_KEY_SNAPSHOT_LOCATION;
 use databend_storages_common_table_meta::table::OPT_KEY_SOURCE_TABLE_ID;
@@ -274,29 +273,11 @@ impl FuseTable {
             ));
         }
 
-        let location = files[0].clone();
-        let reader = MetaReaders::table_snapshot_reader(self.get_operator());
-        let ver = TableMetaLocationGenerator::snapshot_version(location.as_str());
-        let load_params = LoadParams {
-            location,
-            len_hint: None,
-            ver,
-            put_cache: false,
+        let Some(location) = self.snapshot_loc().await? else {
+            return Err(ErrorCode::TableHistoricalDataNotFound("No historical data"));
         };
-        let snapshot = reader.read(&load_params).await?;
-        // Take the prev snapshot as base snapshot to avoid get orphan snapshot.
-        let prev = snapshot.prev_snapshot_id;
-        match prev {
-            Some((id, v)) => {
-                let new_loc = self
-                    .meta_location_generator()
-                    .snapshot_location_from_uuid(&id, v)?;
-                Ok((new_loc, files))
-            }
-            None => Err(ErrorCode::TableHistoricalDataNotFound(
-                "No historical data found at given point",
-            )),
-        }
+
+        Ok((location, files))
     }
 
     #[async_backtrace::framed]

tests/sqllogictests/suites/base/09_fuse_engine/09_0008_fuse_optimize_table.test

@@ -102,8 +102,6 @@ query II
 select segment_count,block_count from fuse_snapshot('db_09_0008', 't') limit 2
 ----
 1 1
-4 4
-
 
 
 query I

tests/suites/0_stateless/20+_others/20_0011_purge_before.result

@@ -6,7 +6,7 @@ checking that after purge (by snapshot id) there should be 4 rows left
 true
 checking that there should are 3 snapshots before purge
 true
-checking that after purge (by timestamp) there should be at least 2 snapshots left
+checking that after purge (by timestamp) there should be 1 snapshot left
 true
 checking that after purge (by timestamp) there should be 4 rows left
 true

tests/suites/0_stateless/20+_others/20_0011_purge_before.sh

@@ -52,8 +52,8 @@ TIMEPOINT=$(echo "select timestamp from fuse_snapshot('default', 't20_0011') whe
 
 ## verify
 echo "set data_retention_time_in_days=0; optimize table t20_0011 purge before (TIMESTAMP => '$TIMEPOINT'::TIMESTAMP)" | $BENDSQL_CLIENT_CONNECT
-echo "checking that after purge (by timestamp) there should be at least 2 snapshots left"
-echo "select count(*)>=2  from fuse_snapshot('default', 't20_0011')" | $BENDSQL_CLIENT_CONNECT
+echo "checking that after purge (by timestamp) there should be 1 snapshot left"
+echo "select count(*)=1  from fuse_snapshot('default', 't20_0011')" | $BENDSQL_CLIENT_CONNECT
 echo "checking that after purge (by timestamp) there should be 4 rows left"
 echo "select count(*)=4  from t20_0011" | $BENDSQL_CLIENT_CONNECT