Merge pull request #4 from zhyass/fix_purge

Fix purge with older version bug

Author: jun0315

src/query/service/tests/it/main.rs

@@ -14,6 +14,7 @@
 
 #![allow(clippy::uninlined_format_args)]
 #![feature(thread_local)]
+#![feature(io_error_other)]
 
 extern crate core;
 

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

@@ -16,19 +16,23 @@
 use std::sync::Arc;
 
 use chrono::Duration;
+use chrono::Utc;
 use common_base::base::tokio;
 use common_catalog::table_context::TableContext;
 use common_exception::Result;
 use common_expression::DataBlock;
+use common_storages_factory::Table;
 use common_storages_fuse::io::MetaWriter;
 use common_storages_fuse::io::SegmentWriter;
 use common_storages_fuse::statistics::gen_columns_statistics;
+use common_storages_fuse::statistics::merge_statistics;
 use common_storages_fuse::FuseTable;
 use futures_util::TryStreamExt;
 use storages_common_table_meta::meta::Location;
 use storages_common_table_meta::meta::SegmentInfo;
 use storages_common_table_meta::meta::Statistics;
 use storages_common_table_meta::meta::TableSnapshot;
+use storages_common_table_meta::meta::TableSnapshotV2;
 use storages_common_table_meta::meta::Versioned;
 use uuid::Uuid;
 
@@ -191,6 +195,7 @@ async fn test_fuse_purge_orphan_retention() -> Result<()> {
         fuse_table,
         segment_locations.clone(),
         Some(new_timestamp),
+        TableSnapshot::VERSION,
     )
     .await?;
 
@@ -200,6 +205,7 @@ async fn test_fuse_purge_orphan_retention() -> Result<()> {
         fuse_table,
         segment_locations.clone(),
         Some(new_timestamp),
+        TableSnapshot::VERSION,
     )
     .await?;
 
@@ -215,6 +221,7 @@ async fn test_fuse_purge_orphan_retention() -> Result<()> {
             fuse_table,
             segment_locations.clone(),
             Some(new_timestamp),
+            TableSnapshot::VERSION,
         )
         .await?;
     }
@@ -243,29 +250,144 @@ async fn test_fuse_purge_orphan_retention() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_fuse_purge_older_version() -> Result<()> {
+    let fixture = TestFixture::new().await;
+    let ctx = fixture.ctx();
+    fixture.create_default_table().await?;
+
+    let now = Utc::now();
+
+    let schema = TestFixture::default_table_schema();
+    let mut table = fixture.latest_default_table().await?;
+    let mut fuse_table = FuseTable::try_from_table(table.as_ref())?;
+    let location_gen = fuse_table.meta_location_generator();
+    let operator = fuse_table.get_operator();
+
+    {
+        let num_of_segments = 3;
+        let blocks_per_segment = 2;
+        let segments =
+            utils::generate_segments(fuse_table, num_of_segments, blocks_per_segment).await?;
+
+        // create snapshot 0, the format version is 2.
+        let locations = vec![segments[0].0.clone()];
+        let id = Uuid::new_v4();
+        let mut snapshot_0 = TableSnapshotV2::new(
+            id,
+            &None,
+            None,
+            schema.as_ref().clone(),
+            segments[0].1.summary.clone(),
+            locations,
+            None,
+            None,
+        );
+        snapshot_0.timestamp = Some(now - Duration::hours(13));
+
+        let new_snapshot_location = location_gen
+            .snapshot_location_from_uuid(&snapshot_0.snapshot_id, TableSnapshotV2::VERSION)?;
+        utils::write_snapshot_v2(&operator, &new_snapshot_location, &snapshot_0).await?;
+
+        // create snapshot 1, the format version is 3.
+        let mut locations = Vec::with_capacity(2);
+        for i in [1, 0] {
+            locations.push(segments[i].0.clone());
+        }
+        let mut snapshot_1 = TableSnapshot::new(
+            Uuid::new_v4(),
+            &snapshot_0.timestamp,
+            Some((snapshot_0.snapshot_id, TableSnapshotV2::VERSION)),
+            schema.as_ref().clone(),
+            Statistics::default(),
+            locations,
+            None,
+            None,
+        );
+        snapshot_1.timestamp = Some(now - Duration::hours(12));
+        snapshot_1.summary = merge_statistics(&snapshot_0.summary, &segments[1].1.summary)?;
+        let new_snapshot_location = location_gen
+            .snapshot_location_from_uuid(&snapshot_1.snapshot_id, TableSnapshot::VERSION)?;
+        snapshot_1
+            .write_meta(&operator, &new_snapshot_location)
+            .await?;
+
+        // create snapshot 2, the format version is 3.
+        let mut locations = Vec::with_capacity(3);
+        for i in [2, 1, 0] {
+            locations.push(segments[i].0.clone());
+        }
+        let mut snapshot_2 = TableSnapshot::from_previous(&snapshot_1);
+        snapshot_2.segments = locations;
+        snapshot_2.timestamp = Some(now);
+        snapshot_2.summary = merge_statistics(&snapshot_1.summary, &segments[2].1.summary)?;
+        let new_snapshot_location = location_gen
+            .snapshot_location_from_uuid(&snapshot_2.snapshot_id, TableSnapshot::VERSION)?;
+        snapshot_2
+            .write_meta(&operator, &new_snapshot_location)
+            .await?;
+        FuseTable::commit_to_meta_server(
+            ctx.as_ref(),
+            fuse_table.get_table_info(),
+            location_gen,
+            snapshot_2,
+            None,
+            &None,
+            &operator,
+        )
+        .await?;
+    }
+
+    let table_ctx: Arc<dyn TableContext> = ctx.clone();
+    table = fixture.latest_default_table().await?;
+    fuse_table = FuseTable::try_from_table(table.as_ref())?;
+    fuse_table.do_purge(&table_ctx, true).await?;
+
+    let expected_num_of_snapshot = 1;
+    let expected_num_of_segment = 3;
+    let expected_num_of_blocks = 6;
+    let expected_num_of_index = expected_num_of_blocks;
+    check_data_dir(
+        &fixture,
+        "do_gc: retention period is 0",
+        expected_num_of_snapshot,
+        0,
+        expected_num_of_segment,
+        expected_num_of_blocks,
+        expected_num_of_index,
+        Some(()),
+        None,
+    )
+    .await?;
+    Ok(())
+}
+
 mod utils {
+    use std::io::Error;
     use std::sync::Arc;
 
     use chrono::DateTime;
     use chrono::Utc;
     use common_storages_factory::Table;
     use common_storages_fuse::io::MetaWriter;
     use common_storages_fuse::FuseStorageFormat;
+    use opendal::Operator;
 
     use super::*;
 
     pub async fn generate_snapshot_with_segments(
         fuse_table: &FuseTable,
         segment_locations: Vec<Location>,
         time_stamp: Option<DateTime<Utc>>,
+        version: u64,
     ) -> Result<String> {
         let current_snapshot = fuse_table.read_table_snapshot().await?.unwrap();
         let operator = fuse_table.get_operator();
         let location_gen = fuse_table.meta_location_generator();
         let mut new_snapshot = TableSnapshot::from_previous(current_snapshot.as_ref());
         new_snapshot.segments = segment_locations;
-        let new_snapshot_location = location_gen
-            .snapshot_location_from_uuid(&new_snapshot.snapshot_id, TableSnapshot::VERSION)?;
+        let new_snapshot_location =
+            location_gen.snapshot_location_from_uuid(&new_snapshot.snapshot_id, version)?;
         if let Some(ts) = time_stamp {
             new_snapshot.timestamp = Some(ts)
         }
@@ -320,4 +442,14 @@ mod utils {
         let segment_location = segment_writer.write_segment_no_cache(&segment_info).await?;
         Ok((segment_location, segment_info))
     }
+
+    pub async fn write_snapshot_v2(
+        data_accessor: &Operator,
+        location: &str,
+        meta: &TableSnapshotV2,
+    ) -> Result<()> {
+        let bs = serde_json::to_vec(&meta).map_err(Error::other)?;
+        data_accessor.write(location, bs).await?;
+        Ok(())
+    }
 }

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

@@ -18,6 +18,7 @@ mod compression;
 /// Re-exports meta data structures of current version, i.e. v1
 mod current;
 mod format;
+mod utils;
 mod v0;
 mod v1;
 mod v2;
@@ -37,6 +38,7 @@ pub use statistics::Location;
 pub use statistics::SnapshotId;
 pub use statistics::Statistics;
 pub use statistics::StatisticsOfColumns;
+pub use utils::*;
 pub use versions::testify_version;
 pub use versions::SegmentInfoVersion;
 pub use versions::SnapshotVersion;

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

@@ -0,0 +1,49 @@
+//  Copyright 2023 Datafuse Labs.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+use std::ops::Add;
+
+use chrono::DateTime;
+use chrono::Datelike;
+use chrono::TimeZone;
+use chrono::Timelike;
+use chrono::Utc;
+
+pub fn trim_timestamp_to_micro_second(ts: DateTime<Utc>) -> DateTime<Utc> {
+    Utc.with_ymd_and_hms(
+        ts.year(),
+        ts.month(),
+        ts.day(),
+        ts.hour(),
+        ts.minute(),
+        ts.second(),
+    )
+    .unwrap()
+    .with_nanosecond(ts.timestamp_subsec_micros() * 1_000)
+    .unwrap()
+}
+
+pub fn monotonically_increased_timestamp(
+    timestamp: DateTime<Utc>,
+    previous_timestamp: &Option<DateTime<Utc>>,
+) -> DateTime<Utc> {
+    if let Some(prev_instant) = previous_timestamp {
+        // timestamp of the snapshot should always larger than the previous one's
+        if prev_instant > &timestamp {
+            // if local time is smaller, use the timestamp of previous snapshot, plus 1 ms
+            return prev_instant.add(chrono::Duration::milliseconds(1));
+        }
+    }
+    timestamp
+}

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

@@ -12,8 +12,6 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
-use std::ops::Add;
-
 use chrono::DateTime;
 use chrono::Utc;
 use common_base::base::uuid::Uuid;
@@ -22,7 +20,9 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use super::super::v0::statistics::Statistics;
+use crate::meta::monotonically_increased_timestamp;
 use crate::meta::statistics::FormatVersion;
+use crate::meta::trim_timestamp_to_micro_second;
 use crate::meta::ClusterKey;
 use crate::meta::Location;
 use crate::meta::SnapshotId;
@@ -74,10 +74,10 @@ impl TableSnapshot {
     ) -> Self {
         let now = Utc::now();
         // make snapshot timestamp monotonically increased
-        let adjusted_timestamp = util::monotonically_increased_timestamp(now, prev_timestamp);
+        let adjusted_timestamp = monotonically_increased_timestamp(now, prev_timestamp);
 
         // trim timestamp to micro seconds
-        let trimmed_timestamp = util::trim_timestamp_to_micro_second(adjusted_timestamp);
+        let trimmed_timestamp = trim_timestamp_to_micro_second(adjusted_timestamp);
         let timestamp = Some(trimmed_timestamp);
 
         Self {
@@ -163,40 +163,3 @@ impl From<&TableSnapshot> for TableSnapshotLite {
         }
     }
 }
-
-mod util {
-    use chrono::DateTime;
-    use chrono::Datelike;
-    use chrono::TimeZone;
-    use chrono::Timelike;
-    use chrono::Utc;
-
-    use super::*;
-    pub fn trim_timestamp_to_micro_second(ts: DateTime<Utc>) -> DateTime<Utc> {
-        Utc.with_ymd_and_hms(
-            ts.year(),
-            ts.month(),
-            ts.day(),
-            ts.hour(),
-            ts.minute(),
-            ts.second(),
-        )
-        .unwrap()
-        .with_nanosecond(ts.timestamp_subsec_micros() * 1_000)
-        .unwrap()
-    }
-
-    pub fn monotonically_increased_timestamp(
-        timestamp: DateTime<Utc>,
-        previous_timestamp: &Option<DateTime<Utc>>,
-    ) -> DateTime<Utc> {
-        if let Some(prev_instant) = previous_timestamp {
-            // timestamp of the snapshot should always larger than the previous one's
-            if prev_instant > &timestamp {
-                // if local time is smaller, use the timestamp of previous snapshot, plus 1 ms
-                return prev_instant.add(chrono::Duration::milliseconds(1));
-            }
-        }
-        timestamp
-    }
-}