Merge pull request #9108 from dantengsky/fix-orphan-snapshots

fix: collect orphan snapshots

Author: BohuTANG

Cargo.lock

@@ -194,7 +194,7 @@ pub trait Table: Sync + Send {
         Ok(())
     }
 
-    async fn optimize(&self, ctx: Arc<dyn TableContext>, keep_last_snapshot: bool) -> Result<()> {
+    async fn purge(&self, ctx: Arc<dyn TableContext>, keep_last_snapshot: bool) -> Result<()> {
         let (_, _) = (ctx, keep_last_snapshot);
 
         Ok(())

src/query/catalog/src/table.rs

@@ -113,7 +113,7 @@ impl Interpreter for OptimizeTableInterpreter {
         }
 
         if do_purge {
-            table.optimize(self.ctx.clone(), true).await?;
+            table.purge(self.ctx.clone(), true).await?;
         }
 
         Ok(PipelineBuildResult::create())

src/query/service/src/interpreters/interpreter_table_optimize.rs

@@ -0,0 +1,117 @@
+// 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 common_ast::ast::Engine;
+use common_base::base::tokio;
+use common_sql::plans::AlterTableClusterKeyPlan;
+use common_sql::plans::CreateTablePlanV2;
+use common_sql::plans::DropTableClusterKeyPlan;
+use common_storages_fuse::io::MetaReaders;
+use common_storages_fuse::FuseTable;
+use common_storages_table_meta::table::OPT_KEY_DATABASE_ID;
+use common_storages_table_meta::table::OPT_KEY_SNAPSHOT_LOCATION;
+use databend_query::interpreters::AlterTableClusterKeyInterpreter;
+use databend_query::interpreters::CreateTableInterpreterV2;
+use databend_query::interpreters::DropTableClusterKeyInterpreter;
+use databend_query::interpreters::Interpreter;
+
+use crate::storages::fuse::table_test_fixture::TestFixture;
+
+#[tokio::test]
+async fn test_fuse_alter_table_cluster_key() -> common_exception::Result<()> {
+    let fixture = TestFixture::new().await;
+    let ctx = fixture.ctx();
+
+    let create_table_plan = CreateTablePlanV2 {
+        if_not_exists: false,
+        tenant: fixture.default_tenant(),
+        catalog: fixture.default_catalog_name(),
+        database: fixture.default_db_name(),
+        table: fixture.default_table_name(),
+        schema: TestFixture::default_schema(),
+        engine: Engine::Fuse,
+        storage_params: None,
+        options: [
+            // database id is required for FUSE
+            (OPT_KEY_DATABASE_ID.to_owned(), "1".to_owned()),
+        ]
+        .into(),
+        field_default_exprs: vec![],
+        field_comments: vec![],
+        as_select: None,
+        cluster_key: None,
+    };
+
+    // create test table
+    let interpreter = CreateTableInterpreterV2::try_create(ctx.clone(), create_table_plan)?;
+    interpreter.execute(ctx.clone()).await?;
+
+    // add cluster key
+    let alter_table_cluster_key_plan = AlterTableClusterKeyPlan {
+        tenant: fixture.default_tenant(),
+        catalog: fixture.default_catalog_name(),
+        database: fixture.default_db_name(),
+        table: fixture.default_table_name(),
+        cluster_keys: vec!["id".to_string()],
+    };
+    let interpreter =
+        AlterTableClusterKeyInterpreter::try_create(ctx.clone(), alter_table_cluster_key_plan)?;
+    interpreter.execute(ctx.clone()).await?;
+
+    let table = fixture.latest_default_table().await?;
+    let fuse_table = FuseTable::try_from_table(table.as_ref())?;
+    let table_info = table.get_table_info();
+    assert_eq!(table_info.meta.cluster_keys, vec!["(id)".to_string()]);
+    assert_eq!(table_info.meta.default_cluster_key_id, Some(0));
+
+    let snapshot_loc = table
+        .get_table_info()
+        .options()
+        .get(OPT_KEY_SNAPSHOT_LOCATION)
+        .unwrap();
+    let reader = MetaReaders::table_snapshot_reader(fuse_table.get_operator());
+    let snapshot = reader.read(snapshot_loc.as_str(), None, 1).await?;
+    let expected = Some((0, "(id)".to_string()));
+    assert_eq!(snapshot.cluster_key_meta, expected);
+
+    // drop cluster key
+    let drop_table_cluster_key_plan = DropTableClusterKeyPlan {
+        tenant: fixture.default_tenant(),
+        catalog: fixture.default_catalog_name(),
+        database: fixture.default_db_name(),
+        table: fixture.default_table_name(),
+    };
+    let interpreter =
+        DropTableClusterKeyInterpreter::try_create(ctx.clone(), drop_table_cluster_key_plan)?;
+    interpreter.execute(ctx.clone()).await?;
+
+    let table = fixture.latest_default_table().await?;
+    let fuse_table = FuseTable::try_from_table(table.as_ref())?;
+    let table_info = table.get_table_info();
+    assert_eq!(table_info.meta.default_cluster_key, None);
+    assert_eq!(table_info.meta.default_cluster_key_id, None);
+
+    let snapshot_loc = table
+        .get_table_info()
+        .options()
+        .get(OPT_KEY_SNAPSHOT_LOCATION)
+        .unwrap();
+    let reader = MetaReaders::table_snapshot_reader(fuse_table.get_operator());
+    let snapshot = reader.read(snapshot_loc.as_str(), None, 1).await?;
+    let expected = None;
+    assert_eq!(snapshot.cluster_key_meta, expected);
+
+    Ok(())
+}

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

@@ -232,7 +232,6 @@ async fn test_abort_on_error() -> Result<()> {
             )) {
                 let entry = entry.unwrap();
                 if entry.file_type().is_file() {
-                    eprintln!("{:?}", entry);
                     ss_count += 1;
                 }
             }