chore: reinfe log for transaction commit (#18435)

* fm unused code

* refine log

* refine error message

* fix ut

Author: SkyFan2002

src/meta/api/src/schema_api_impl.rs

@@ -39,6 +39,7 @@ use databend_common_meta_app::app_error::DatabaseAlreadyExists;
 use databend_common_meta_app::app_error::DropDbWithDropTime;
 use databend_common_meta_app::app_error::DropTableWithDropTime;
 use databend_common_meta_app::app_error::DuplicatedIndexColumnId;
+use databend_common_meta_app::app_error::DuplicatedUpsertFiles;
 use databend_common_meta_app::app_error::IndexColumnIdNotFound;
 use databend_common_meta_app::app_error::MultiStmtTxnCommitFailed;
 use databend_common_meta_app::app_error::StreamAlreadyExists;
@@ -2205,6 +2206,12 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
         // in each function. In this case there is chance that some `TableCopiedFileInfo` may not be
         // removed in `remove_table_copied_files`, but these data can be purged in case of expire time.
 
+        let insert_if_not_exists_table_ids = copied_files
+            .iter()
+            .filter(|(_, req)| req.insert_if_not_exists)
+            .map(|(table_id, _)| *table_id)
+            .collect::<Vec<_>>();
+
         for (table_id, req) in copied_files {
             let tbid = TableId { table_id };
 
@@ -2300,10 +2307,20 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
         }
 
         if mismatched_tbs.is_empty() {
-            // if all table version does match, but tx failed, we don't know why, just return error
-            Err(KVAppError::AppError(AppError::from(
-                MultiStmtTxnCommitFailed::new("update_multi_table_meta"),
-            )))
+            if !insert_if_not_exists_table_ids.is_empty() {
+                // If insert_if_not_exists is true and transaction failed, it's likely due to duplicated files
+                Err(KVAppError::AppError(AppError::from(
+                    DuplicatedUpsertFiles::new(
+                        insert_if_not_exists_table_ids,
+                        "update_multi_table_meta",
+                    ),
+                )))
+            } else {
+                // if all table version does match, but tx failed, we don't know why, just return error
+                Err(KVAppError::AppError(AppError::from(
+                    MultiStmtTxnCommitFailed::new("update_multi_table_meta"),
+                )))
+            }
         } else {
             // up layer will retry
             Ok(Err(mismatched_tbs))

src/meta/api/src/schema_api_test_suite.rs

@@ -2792,7 +2792,7 @@ impl SchemaApiTestSuite {
                     .await;
                 let err = result.unwrap_err();
                 let err = ErrorCode::from(err);
-                assert_eq!(ErrorCode::UNRESOLVABLE_CONFLICT, err.code());
+                assert_eq!(ErrorCode::DUPLICATED_UPSERT_FILES, err.code());
             }
 
             info!("--- update table meta: virtual column too many");
@@ -7762,7 +7762,7 @@ impl SchemaApiTestSuite {
             let result = mt.update_multi_table_meta(req).await;
             let err = result.unwrap_err();
             let err = ErrorCode::from(err);
-            assert_eq!(ErrorCode::UNRESOLVABLE_CONFLICT, err.code());
+            assert_eq!(ErrorCode::DUPLICATED_UPSERT_FILES, err.code());
 
             let req = GetTableCopiedFileReq {
                 table_id,

src/meta/app/src/app_error.rs

@@ -361,14 +361,14 @@ impl crate::app_error::UpdateStreamMetasFailed {
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
-#[error("DuplicatedUpsertFiles: {table_id} , in operation `{context}`")]
+#[error("DuplicatedUpsertFiles: {table_id:?} , in operation `{context}`")]
 pub struct DuplicatedUpsertFiles {
-    table_id: u64,
+    table_id: Vec<u64>,
     context: String,
 }
 
 impl DuplicatedUpsertFiles {
-    pub fn new(table_id: u64, context: impl Into<String>) -> Self {
+    pub fn new(table_id: Vec<u64>, context: impl Into<String>) -> Self {
         DuplicatedUpsertFiles {
             table_id,
             context: context.into(),

src/query/catalog/src/database.rs

@@ -34,8 +34,6 @@ use databend_common_meta_app::schema::TableInfo;
 use databend_common_meta_app::schema::TruncateTableReply;
 use databend_common_meta_app::schema::TruncateTableReq;
 use databend_common_meta_app::schema::UndropTableReq;
-use databend_common_meta_app::schema::UpdateMultiTableMetaReq;
-use databend_common_meta_app::schema::UpdateMultiTableMetaResult;
 use databend_common_meta_app::schema::UpsertTableOptionReply;
 use databend_common_meta_app::schema::UpsertTableOptionReq;
 use databend_common_meta_app::tenant::Tenant;
@@ -218,15 +216,4 @@ pub trait Database: DynClone + Sync + Send {
             self.name()
         )))
     }
-
-    #[async_backtrace::framed]
-    async fn retryable_update_multi_table_meta(
-        &self,
-        _req: UpdateMultiTableMetaReq,
-    ) -> Result<UpdateMultiTableMetaResult> {
-        Err(ErrorCode::Unimplemented(format!(
-            "UnImplement retryable_update_multi_table_meta in {} Database",
-            self.name()
-        )))
-    }
 }

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

@@ -697,9 +697,24 @@ impl Catalog for MutableCatalog {
             }
         }
 
+        let table_updates: Vec<String> = req
+            .update_table_metas
+            .iter()
+            .map(|(update_req, _)| {
+                format!("table_id={}, seq={}", update_req.table_id, update_req.seq)
+            })
+            .collect();
+
+        let stream_updates: Vec<String> = req
+            .update_stream_metas
+            .iter()
+            .map(|stream_req| format!("stream_id={}, seq={}", stream_req.stream_id, stream_req.seq))
+            .collect();
+
         info!(
-            "[CATALOG] Updating multiple table metadata: table_count={}",
-            req.update_table_metas.len()
+            "[CATALOG] Updating multiple table metadata: table_updates=[{}], stream_updates=[{}]",
+            table_updates.join("; "),
+            stream_updates.join("; ")
         );
         let begin = Instant::now();
         let res = self.ctx.meta.update_multi_table_meta(req).await;

src/query/service/src/databases/default/default_database.rs

@@ -42,8 +42,6 @@ use databend_common_meta_app::schema::TableInfo;
 use databend_common_meta_app::schema::TruncateTableReply;
 use databend_common_meta_app::schema::TruncateTableReq;
 use databend_common_meta_app::schema::UndropTableReq;
-use databend_common_meta_app::schema::UpdateMultiTableMetaReq;
-use databend_common_meta_app::schema::UpdateMultiTableMetaResult;
 use databend_common_meta_app::schema::UpsertTableOptionReply;
 use databend_common_meta_app::schema::UpsertTableOptionReq;
 
@@ -291,13 +289,4 @@ impl Database for DefaultDatabase {
         let res = self.ctx.meta.truncate_table(req).await?;
         Ok(res)
     }
-
-    #[async_backtrace::framed]
-    async fn retryable_update_multi_table_meta(
-        &self,
-        req: UpdateMultiTableMetaReq,
-    ) -> Result<UpdateMultiTableMetaResult> {
-        let res = self.ctx.meta.update_multi_table_meta(req).await?;
-        Ok(res)
-    }
 }