chore: refine error message in transaction retry and add more log (#18528)

* refine error message

* add log

* add log

Author: SkyFan2002

src/query/ee/src/stream/handler.rs

@@ -119,7 +119,7 @@ impl StreamHandler for RealStreamHandler {
             .get(OPT_KEY_DATABASE_ID)
             .ok_or_else(|| {
                 ErrorCode::Internal(format!(
-                    "Invalid fuse table, table option {} not found",
+                    "Invalid fuse table, table option {} not found when creating stream",
                     OPT_KEY_DATABASE_ID
                 ))
             })?;

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

@@ -713,15 +713,17 @@ impl Catalog for MutableCatalog {
             .collect();
 
         info!(
-            "[CATALOG] Updating multiple table metadata: table_updates=[{}], stream_updates=[{}]",
+            "[CATALOG] Updating multiple table metadata: table_updates=[{}], stream_updates=[{}], req={:?}",
             table_updates.join("; "),
-            stream_updates.join("; ")
+            stream_updates.join("; "),
+            req
         );
         let begin = Instant::now();
         let res = self.ctx.meta.update_multi_table_meta(req).await;
         info!(
-            "[CATALOG] Multiple table metadata update completed: elapsed_time={:?}",
-            begin.elapsed()
+            "[CATALOG] Multiple table metadata update completed: elapsed_time={:?}, result={:?}",
+            begin.elapsed(),
+            res
         );
         Ok(res?)
     }

src/query/service/src/interpreters/common/stream.rs

@@ -70,7 +70,7 @@ pub async fn dml_build_update_stream_req(
                 .get(OPT_KEY_DATABASE_ID)
                 .ok_or_else(|| {
                     ErrorCode::Internal(format!(
-                        "Invalid fuse table, table option {} not found",
+                        "Invalid fuse table, table option {} not found when building update stream req",
                         OPT_KEY_DATABASE_ID
                     ))
                 })?;

src/query/storages/common/table_meta/src/meta/utils.rs

@@ -160,7 +160,7 @@ pub fn parse_storage_prefix(options: &BTreeMap<String, String>, table_id: u64) -
 
     let db_id = options.get(OPT_KEY_DATABASE_ID).ok_or_else(|| {
         ErrorCode::Internal(format!(
-            "Invalid fuse table, table option {} not found",
+            "Invalid fuse table, table option {} not found when parsing storage prefix",
             OPT_KEY_DATABASE_ID
         ))
     })?;

src/query/storages/fuse/src/retry/commit.rs

@@ -28,6 +28,7 @@ use databend_storages_common_cache::Table;
 use databend_storages_common_cache::TableSnapshot;
 use databend_storages_common_table_meta::meta::Versioned;
 use databend_storages_common_table_meta::readers::snapshot_reader::TableSnapshotAccessor;
+use log::info;
 
 use super::diff::SegmentsDiff;
 use crate::operations::set_backoff;
@@ -68,8 +69,18 @@ async fn try_rebuild_req(
     req: &mut UpdateMultiTableMetaReq,
     update_failed_tbls: Vec<(u64, u64, TableMeta)>,
 ) -> Result<()> {
+    info!(
+        "try_rebuild_req: update_failed_tbls={:?}",
+        update_failed_tbls
+    );
     let txn_mgr = ctx.txn_mgr();
     for (tid, seq, table_meta) in update_failed_tbls {
+        if table_meta.engine == "STREAM" {
+            return Err(ErrorCode::UnresolvableConflict(format!(
+                "Concurrent transaction commit failed. Stream table {} has unresolvable conflicts.",
+                tid
+            )));
+        }
         let latest_table = FuseTable::from_table_meta(tid, seq, table_meta)?;
         let default_cluster_key_id = latest_table.cluster_key_id();
         let latest_snapshot = latest_table.read_table_snapshot().await?;

src/query/storages/stream/src/stream_table.rs

@@ -290,7 +290,7 @@ impl StreamTable {
                     .get(OPT_KEY_DATABASE_ID)
                     .ok_or_else(|| {
                         ErrorCode::Internal(format!(
-                            "Invalid fuse table, table option {} not found",
+                            "Invalid fuse table, table option {} not found when getting source database id",
                             OPT_KEY_DATABASE_ID
                         ))
                     })?