add truncate_table in catalog to truncate all the copied files info in meta service

Author: lichuang

src/query/catalog/src/catalog.rs

@@ -35,6 +35,8 @@ use common_meta_app::schema::RenameTableReq;
 use common_meta_app::schema::TableIdent;
 use common_meta_app::schema::TableInfo;
 use common_meta_app::schema::TableMeta;
+use common_meta_app::schema::TruncateTableReply;
+use common_meta_app::schema::TruncateTableReq;
 use common_meta_app::schema::UndropDatabaseReply;
 use common_meta_app::schema::UndropDatabaseReq;
 use common_meta_app::schema::UndropTableReply;
@@ -166,6 +168,8 @@ pub trait Catalog: DynClone + Send + Sync {
         req: UpsertTableCopiedFileReq,
     ) -> Result<UpsertTableCopiedFileReply>;
 
+    async fn truncate_table(&self, req: TruncateTableReq) -> Result<TruncateTableReply>;
+
     /// Table function
 
     // Get function by name.

src/query/service/src/catalogs/default/database_catalog.rs

@@ -34,6 +34,8 @@ use common_meta_app::schema::RenameTableReq;
 use common_meta_app::schema::TableIdent;
 use common_meta_app::schema::TableInfo;
 use common_meta_app::schema::TableMeta;
+use common_meta_app::schema::TruncateTableReply;
+use common_meta_app::schema::TruncateTableReq;
 use common_meta_app::schema::UndropDatabaseReply;
 use common_meta_app::schema::UndropDatabaseReq;
 use common_meta_app::schema::UndropTableReply;
@@ -457,6 +459,10 @@ impl Catalog for DatabaseCatalog {
             .await
     }
 
+    async fn truncate_table(&self, req: TruncateTableReq) -> Result<TruncateTableReply> {
+        self.mutable_catalog.truncate_table(req).await
+    }
+
     async fn upsert_table_option(
         &self,
         req: UpsertTableOptionReq,

src/query/service/src/catalogs/default/immutable_catalog.rs

@@ -34,6 +34,8 @@ use common_meta_app::schema::RenameTableReq;
 use common_meta_app::schema::TableIdent;
 use common_meta_app::schema::TableInfo;
 use common_meta_app::schema::TableMeta;
+use common_meta_app::schema::TruncateTableReply;
+use common_meta_app::schema::TruncateTableReq;
 use common_meta_app::schema::UndropDatabaseReply;
 use common_meta_app::schema::UndropDatabaseReq;
 use common_meta_app::schema::UndropTableReply;
@@ -216,6 +218,13 @@ impl Catalog for ImmutableCatalog {
         )))
     }
 
+    async fn truncate_table(&self, req: TruncateTableReq) -> Result<TruncateTableReply> {
+        Err(ErrorCode::UnImplement(format!(
+            "truncate_table not allowed for system database {:?}",
+            req
+        )))
+    }
+
     async fn upsert_table_option(
         &self,
         req: UpsertTableOptionReq,

src/query/service/src/catalogs/default/mutable_catalog.rs

@@ -42,6 +42,8 @@ use common_meta_app::schema::RenameTableReq;
 use common_meta_app::schema::TableIdent;
 use common_meta_app::schema::TableInfo;
 use common_meta_app::schema::TableMeta;
+use common_meta_app::schema::TruncateTableReply;
+use common_meta_app::schema::TruncateTableReq;
 use common_meta_app::schema::UndropDatabaseReply;
 use common_meta_app::schema::UndropDatabaseReq;
 use common_meta_app::schema::UndropTableReply;
@@ -332,6 +334,11 @@ impl Catalog for MutableCatalog {
         Ok(res)
     }
 
+    async fn truncate_table(&self, req: TruncateTableReq) -> Result<TruncateTableReply> {
+        let res = self.ctx.meta.truncate_table(req).await?;
+        Ok(res)
+    }
+
     async fn count_tables(&self, req: CountTablesReq) -> Result<CountTablesReply> {
         let res = self.ctx.meta.count_tables(req).await?;
         Ok(res)

src/query/service/src/interpreters/interpreter_copy_v2.rs

@@ -76,6 +76,7 @@ impl CopyInterpreterV2 {
         let mut file_map = BTreeMap::new();
 
         if !force {
+            // if force is false, copy only the files that unmatch to the meta copied files info.
             let resp = catalog.get_table_copied_file_info(req).await?;
             for file in files.iter() {
                 let stage_file = list_files(&self.ctx, &table_info.stage_info, file, "").await?;
@@ -88,12 +89,17 @@ impl CopyInterpreterV2 {
                 if let Some(file_info) = resp.file_info.get(file) {
                     // No need to copy the file again if etag is_some and match.
                     if stage_file.etag.is_some() && stage_file.etag == file_info.etag {
+                        tracing::warn!("ignore copy file {:?} matched by etag", file);
                         continue;
                     }
 
                     if file_info.content_length == stage_file.size
                         && file_info.last_modified == Some(stage_file.last_modified)
                     {
+                        tracing::warn!(
+                            "ignore copy file {:?} matched by content_length and last_modified",
+                            file
+                        );
                         continue;
                     }
                 }
@@ -106,6 +112,7 @@ impl CopyInterpreterV2 {
                 });
             }
         } else {
+            // if force is true, copy all the file.
             for file in files.iter() {
                 let stage_file = list_files(&self.ctx, &table_info.stage_info, file, "").await?;
                 if stage_file.is_empty() {

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

@@ -19,6 +19,7 @@ use common_exception::Result;
 use common_fuse_meta::meta::TableSnapshot;
 use common_fuse_meta::meta::Versioned;
 use common_meta_app::schema::TableStatistics;
+use common_meta_app::schema::TruncateTableReq;
 use common_meta_app::schema::UpdateTableMetaReq;
 use common_meta_types::MatchSeq;
 use uuid::Uuid;
@@ -69,13 +70,19 @@ impl FuseTable {
 
             let table_id = self.table_info.ident.table_id;
             let table_version = self.table_info.ident.seq;
-            ctx.get_catalog(catalog_name)?
+            let catalog = ctx.get_catalog(catalog_name)?;
+
+            catalog
                 .update_table_meta(UpdateTableMetaReq {
                     table_id,
                     seq: MatchSeq::Exact(table_version),
                     new_table_meta,
                 })
                 .await?;
+
+            catalog
+                .truncate_table(TruncateTableReq { table_id })
+                .await?;
         }
 
         Ok(())

src/query/storages/hive/src/hive_catalog.rs

@@ -44,6 +44,8 @@ use common_meta_app::schema::RenameTableReq;
 use common_meta_app::schema::TableIdent;
 use common_meta_app::schema::TableInfo;
 use common_meta_app::schema::TableMeta;
+use common_meta_app::schema::TruncateTableReply;
+use common_meta_app::schema::TruncateTableReq;
 use common_meta_app::schema::UndropDatabaseReply;
 use common_meta_app::schema::UndropDatabaseReq;
 use common_meta_app::schema::UndropTableReply;
@@ -350,6 +352,10 @@ impl Catalog for HiveCatalog {
         unimplemented!()
     }
 
+    async fn truncate_table(&self, _req: TruncateTableReq) -> Result<TruncateTableReply> {
+        unimplemented!()
+    }
+
     async fn count_tables(&self, _req: CountTablesReq) -> Result<CountTablesReply> {
         unimplemented!()
     }