You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: gc.rs
    path of file: ./repos/databend/src/query/service/tests/it/storages/fuse/operations
    
    the code of the file until where you have to start completion: // Copyright 2022 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::sync::Arc;

use chrono::Duration;
use chrono::Utc;
use databend_common_base::base::tokio;
use databend_common_catalog::table_context::TableContext;
use databend_common_exception::Result;
use databend_common_storages_fuse::io::MetaWriter;
use databend_common_storages_fuse::FuseTable;
use databend_query::test_kits::*;
use databend_storages_common_table_meta::meta::Location;
use databend_storages_common_table_meta::meta::TableSnapshot;
use databend_storages_common_table_meta::meta::Versioned;
use uuid::Uuid;

use crate::storages::fuse::operations::mutation::compact_segment;

#[tokio::test(flavor = "multi_thread")]
async fn test_fuse_purge_normal_case() -> Result<()> {
    let fixture = TestFixture::setup().await?;
    fixture.create_default_database().await?;
    fixture.create_default_table().await?;

    let ctx = fixture.new_query_ctx().await?;

    // ingests some test data
    append_sample_data(1, &fixture).await?;

    // do_gc
    let table = fixture.latest_default_table().await?;
    let fuse_table = FuseTable::try_from_table(table.as_ref())?;
    let snapshot_files = fuse_table.list_snapshot_files().await?;
    let keep_last_snapshot = true;
    let table_ctx: Arc<dyn TableContext> = ctx.clone();
    fuse_table
        .do_purge(&table_ctx, snapshot_files, None, keep_last_snapshot, false)
        .await?;

    let expected_num_of_snapshot = 1;
    check_data_dir(
        &fixture,
        "do_gc: there should be 1 snapshot, 0 segment/block",
        expected_num_of_snapshot,
        0, // 0 snapshot statistic
        1, // 1 segments
        1, // 1 blocks
        1, // 1 index
        Some(()),
        None,
    )
    .await?;

    Ok(())
}

#[tokio::test(flavor = "multi_thread")]
async fn test_fuse_purge_normal_orphan_snapshot() -> Result<()> {
    let fixture = TestFixture::setup().await?;
    fixture.create_default_database().await?;
    fixture.create_default_table().await?;

    let ctx = fixture.new_query_ctx().await?;

    // ingests some test data
    append_sample_data(1, &fixture).await?;

    let table = fixture.latest_default_table().await?;
    let fuse_table = FuseTable::try_from_table(table.as_ref())?;

    // create orphan snapshot, its timestamp is larger than the current one
    {
        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 orphan_snapshot_id = Uuid::new_v4();
  
    
async fn test_fuse_purge_older_version() -> Result<()> {
    let fixture = TestFixture::setup().await?;
    fixture.create_default_database().await?;
    fixture.create_normal_table().await?;

    generate_snapshots(&fixture).await?;
    let ctx = fixture.new_query_ctx().await?;
    let table_ctx: Arc<dyn TableContext> = ctx.clone();
    let now = Utc::now();

    // navigate to time point, snapshot 0 is purged.
    {
        let latest_table = fixture.latest_default_table().await?;
        let fuse_table = FuseTable::try_from_table(latest_table.as_ref())?;
        let snapshot_files = fuse_table.list_snapshot_files().await?;
        let time_point = now - Duration::hours(12);
        let snapshot_loc = fuse_table.snapshot_loc().await?.unwrap();
        let table = fuse_table
            .navigate_to_time_point(snapshot_loc, time_point)
            .await?;
        let keep_last_snapshot = true;
        table
            .do_purge(&table_ctx, snapshot_files, None, keep_last_snapshot, false)
            .await?;

        let expected_num_of_snapshot = 2;
        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: navigate to time point",
            expected_num_of_snapshot,
            0,
            expected_num_of_segment,
            expected_num_of_blocks,
            expected_num_of_index,
            Some(()),
            None,
        )
        .await?;
    }

    // ingests some test data
    append_sample_data(1, &fixture).await?;

    // Do compact segment, generate a new snapshot.
    {
        let table = fixture.latest_default_table().await?;
        compact_segment(ctx.clone(), &table).await?;
        check_data_dir(&fixture, "", 4, 0, 5, 7, 7, Some(()), None).await?;
    }

    let table = fixture.latest_default_table().await?;
    let fuse_table = FuseTable::try_from_table(table.as_ref())?;
    // base snapshot is root. do purge.
    {
        let snapshot_files = fuse_table.list_snapshot_files().await?;
        fuse_table
            .do_purge(&table_ctx, snapshot_files, None, true, false)
            .await?;

        let expected_num_of_snapshot = 1;
        let expected_num_of_segment = 1;
        let expected_num_of_blocks = 7;
        let expected_num_of_index = expected_num_of_blocks;
        check_data_dir(
            &fixture,
            "do_gc: with older version",
            expected_num_of_snapshot,
            0,
            expected_num_of_segment,
            expected_num_of_blocks,
            expected_num_of_index,
            Some(()),
            None,
        )
        .await?;
    }

    // keep_last_snapshot is false. All of snapshots will be purged.
    {
        let snapshot_files = fuse_table.list_snapshot_files().await?;
        fuse_table
            .do_purge(&table_ctx, snapshot_files, None, false, false)
            .await?;
        let expected_num_of_snapshot = 0;
        let expected_num_of_segment = 0;
        let expected_num_of_blocks = 0;
        let expected_num_of_index = expected_num_of_blocks;
        check_data_dir(
            &fixture,
            "do_gc: purge last snapshot",
            expected_num_of_snapshot,
            0,
            expected_num_of_segment,
            expected_num_of_blocks,
            expected_num_of_index,
            Some(()),
            None,
        )
        .await?;
    }

    Ok(())
}