fix: purge oom (#12950)

* fix purge oom

* add unit test

Author: zhyass

src/query/service/tests/it/storages/fuse/meta/snapshot.rs

@@ -12,9 +12,13 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+use std::collections::HashMap;
 use std::ops::Add;
 
 use common_expression::TableSchema;
+use storages_common_table_meta::meta::testing::StatisticsV0;
+use storages_common_table_meta::meta::testing::TableSnapshotV1;
+use storages_common_table_meta::meta::testing::TableSnapshotV2;
 use storages_common_table_meta::meta::TableSnapshot;
 use uuid::Uuid;
 
@@ -74,3 +78,32 @@ fn snapshot_timestamp_time_skew_tolerance() {
     let prev_ts = prev.timestamp.unwrap();
     assert!(current_ts > prev_ts)
 }
+
+#[test]
+fn test_snapshot_v1_to_v4() {
+    let summary = StatisticsV0 {
+        row_count: 0,
+        block_count: 0,
+        perfect_block_count: 0,
+        uncompressed_byte_size: 0,
+        compressed_byte_size: 0,
+        index_size: 0,
+        col_stats: HashMap::new(),
+    };
+    let v1 = TableSnapshotV1::new(
+        Uuid::new_v4(),
+        &None,
+        None,
+        Default::default(),
+        summary,
+        vec![],
+        None,
+        None,
+    );
+    assert!(v1.timestamp.is_some());
+
+    let v4: TableSnapshot = TableSnapshotV2::from(v1.clone()).into();
+    assert_eq!(v4.format_version, v1.format_version());
+    assert_eq!(v4.snapshot_id, v1.snapshot_id);
+    assert_eq!(v4.timestamp, v1.timestamp);
+}

src/query/storages/common/table-meta/src/meta/mod.rs

@@ -50,6 +50,8 @@ pub use versions::Versioned;
 // - export meta encoding to benchmarking tests
 pub mod testing {
     pub use super::format::MetaEncoding;
+    pub use super::v0::statistics::Statistics as StatisticsV0;
+    pub use super::v1::TableSnapshot as TableSnapshotV1;
     pub use super::v2::SegmentInfo as SegmentInfoV2;
     pub use super::v2::TableSnapshot as TableSnapshotV2;
     pub use super::v3::SegmentInfo as SegmentInfoV3;

src/query/storages/common/table-meta/src/meta/v2/snapshot.rs

@@ -127,7 +127,7 @@ impl From<v1::TableSnapshot> for TableSnapshot {
             // carries the format_version of snapshot being converted.
             format_version: s.format_version,
             snapshot_id: s.snapshot_id,
-            timestamp: None,
+            timestamp: s.timestamp,
             prev_snapshot_id: s.prev_snapshot_id,
             schema,
             summary,

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

@@ -63,6 +63,12 @@ impl FuseTable {
             }
         }
         let root_snapshot_info = root_snapshot_info_op.unwrap();
+        if root_snapshot_info.snapshot_lite.timestamp.is_none() {
+            return Err(ErrorCode::StorageOther(format!(
+                "gc: snapshot timestamp is none, snapshot location: {}",
+                root_snapshot_info.snapshot_location
+            )));
+        }
 
         let snapshots_io = SnapshotsIO::create(ctx.clone(), self.operator.clone());
         let location_gen = self.meta_location_generator();
@@ -116,7 +122,7 @@ impl FuseTable {
             let mut segments_to_be_purged = HashSet::new();
             let mut ts_to_be_purged = HashSet::new();
             for s in snapshots.into_iter() {
-                if s.timestamp >= base_timestamp {
+                if s.timestamp.is_some() && s.timestamp >= base_timestamp {
                     remain_snapshots.push(s);
                     continue;
                 }