From 096dd0f4c91d59d67a294255dcd3179bc781870f Mon Sep 17 00:00:00 2001
From: meiyi <meiyi@selectdb.com>
Date: Wed, 2 Sep 2026 18:41:20 +0800
Subject: [PATCH] [fix](fe) Return transaction insert execution errors (#67386)

Transaction INSERT execution failures raised directly while dispatching fragments bypass ErrorReport. OlapTxnInsertExecutor catches these failures and aborts the subtransaction, but previously left QueryState successful, causing the client to receive Query OK. Set the INSERT error state when no earlier error exists so the failure reaches the client while preserving errors already reported by ErrorReport.
---
 .../CloudGlobalTransactionMgr.java            |  2 +-
 .../insert/OlapGroupCommitInsertExecutor.java | 17 +--------
 .../commands/insert/OlapInsertExecutor.java   |  6 +++-
 .../insert/OlapTxnInsertExecutor.java         |  1 +
 .../transaction/txn_insert_inject_case.groovy | 35 +++++++++++++------
 5 files changed, 33 insertions(+), 28 deletions(-)

diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
index df0e193b0e4eae..1c49b3c11b2447 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
@@ -2681,7 +2681,7 @@ public TransactionState abortSubTxn(long txnId, long subTxnId, long dbId, Set<Lo
         LOG.info("try to abort sub transaction, txnId: {}, subTxnId: {}, dbId: {}, tableIds: {}, subTxnNum: {}", txnId,
                 subTxnId, dbId, tableIds, subTxnNum);
         AbortSubTxnRequest request = AbortSubTxnRequest.newBuilder().setCloudUniqueId(Config.cloud_unique_id)
-                .setTxnId(txnId).setSubTxnId(subTxnId).setDbId(dbId).addAllTableIds(tableIds).setSubTxnNum(subTxnId)
+                .setTxnId(txnId).setSubTxnId(subTxnId).setDbId(dbId).addAllTableIds(tableIds).setSubTxnNum(subTxnNum)
                 .setRequestIp(FrontendOptions.getLocalHostAddressCached()).build();
         AbortSubTxnResponse response = null;
         int retryTime = 0;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapGroupCommitInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapGroupCommitInsertExecutor.java
index 0b4d9026d91be3..5b2fe260b3b0c1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapGroupCommitInsertExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapGroupCommitInsertExecutor.java
@@ -22,8 +22,6 @@
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.Table;
 import org.apache.doris.catalog.TableIf;
-import org.apache.doris.common.Config;
-import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.FeConstants;
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.util.DebugUtil;
@@ -44,8 +42,6 @@
 import org.apache.doris.system.Backend;
 import org.apache.doris.transaction.TransactionStatus;
 
-import com.google.common.base.Strings;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -192,18 +188,7 @@ protected void onFail(Throwable t) {
         String queryId = DebugUtil.printId(ctx.queryId());
         // if any throwable being thrown during insert operation, first we should abort this txn
         LOG.warn("insert [{}] with query id {} failed, url={}", labelName, queryId, coordinator.getTrackingUrl(), t);
-        String firstErrorMsgPart = "";
-        String urlPart = "";
-        if (!Strings.isNullOrEmpty(coordinator.getFirstErrorMsg())) {
-            firstErrorMsgPart = StringUtils.abbreviate(coordinator.getFirstErrorMsg(),
-                    Config.first_error_msg_max_length);
-        }
-        if (!Strings.isNullOrEmpty(coordinator.getTrackingUrl())) {
-            urlPart = coordinator.getTrackingUrl();
-        }
-
-        String finalErrorMsg = InsertUtils.getFinalErrorMsg(errMsg, firstErrorMsgPart, urlPart);
-        ctx.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR, finalErrorMsg);
+        setErrorState();
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
index 3771c415c17105..6cbc9d40c88741 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
@@ -307,6 +307,11 @@ protected void onFail(Throwable t) {
         if (Config.isCloudMode() && SystemInfoService.needRetryWithReplan(t.getMessage())) {
             return;
         }
+        setErrorState();
+        recordLoadJob(ctx.getCurrentUserIdentity());
+    }
+
+    protected void setErrorState() {
         String firstErrorMsgPart = "";
         String urlPart = "";
         if (!Strings.isNullOrEmpty(coordinator.getFirstErrorMsg())) {
@@ -318,7 +323,6 @@ protected void onFail(Throwable t) {
         }
         String finalErrorMsg = InsertUtils.getFinalErrorMsg(errMsg, firstErrorMsgPart, urlPart);
         ctx.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR, finalErrorMsg);
-        recordLoadJob(ctx.getCurrentUserIdentity());
     }
 
     private void recordLoadJob(UserIdentity userIdentity) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapTxnInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapTxnInsertExecutor.java
index aa095d3f91af1e..ad7ee6c067ba31 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapTxnInsertExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapTxnInsertExecutor.java
@@ -87,6 +87,7 @@ protected void onFail(Throwable t) {
         // if any throwable being thrown during insert operation, first we should abort this txn
         LOG.warn("insert [{}] with query id {} failed, url={}", labelName, queryId, coordinator.getTrackingUrl(), t);
         cleanTransaction();
+        setErrorState();
     }
 
     @Override
diff --git a/regression-test/suites/insert_p0/transaction/txn_insert_inject_case.groovy b/regression-test/suites/insert_p0/transaction/txn_insert_inject_case.groovy
index 65556a9130109a..39620e8d4b1e5e 100644
--- a/regression-test/suites/insert_p0/transaction/txn_insert_inject_case.groovy
+++ b/regression-test/suites/insert_p0/transaction/txn_insert_inject_case.groovy
@@ -40,6 +40,7 @@ suite("txn_insert_inject_case", "nonConcurrent") {
         """
     }
     GetDebugPoint().disableDebugPointForAllBEs("FlushToken.submit_flush_error")
+    GetDebugPoint().disableDebugPointForAllBEs("FragmentMgr.exec_plan_fragment.failed")
     sql """insert into ${table}_1 values(1, 2.2, "abc", [], []), (2, 3.3, "xyz", [1], [1, 0]), (null, null, null, [null], [null, 0])  """
     sql """insert into ${table}_2 values(3, 2.2, "abc", [], []), (4, 3.3, "xyz", [1], [1, 0]), (null, null, null, [null], [null, 0])  """
 
@@ -48,20 +49,20 @@ suite("txn_insert_inject_case", "nonConcurrent") {
     (ipList, portList) = GetDebugPoint().getBEHostAndHTTPPort()
     logger.info("be ips: ${ipList}, ports: ${portList}")
 
-    def enableDebugPoint = { ->
+    def enableDebugPoint = { debugPoint ->
         ipList.each { beid, ip ->
-            DebugPoint.enableDebugPoint(ip, portList[beid] as int, NodeType.BE, "FlushToken.submit_flush_error")
+            DebugPoint.enableDebugPoint(ip, portList[beid] as int, NodeType.BE, debugPoint)
         }
     }
 
-    def disableDebugPoint = { ->
+    def disableDebugPoint = { debugPoint ->
         ipList.each { beid, ip ->
-            DebugPoint.disableDebugPoint(ip, portList[beid] as int, NodeType.BE, "FlushToken.submit_flush_error")
+            DebugPoint.disableDebugPoint(ip, portList[beid] as int, NodeType.BE, debugPoint)
         }
     }
 
     try {
-        enableDebugPoint()
+        enableDebugPoint("FlushToken.submit_flush_error")
         sql """ begin """
         try {
             sql """ insert into ${table}_0 select * from ${table}_1; """
@@ -78,10 +79,10 @@ suite("txn_insert_inject_case", "nonConcurrent") {
             assertTrue(e.getMessage().contains("dbug_be_memtable_submit_flush_error"))
         }
 
-        disableDebugPoint()
+        disableDebugPoint("FlushToken.submit_flush_error")
         sql """ insert into ${table}_0 select * from ${table}_1; """
 
-        enableDebugPoint()
+        enableDebugPoint("FlushToken.submit_flush_error")
         try {
             sql """ insert into ${table}_0 select * from ${table}_1; """
             assertTrue(false, "insert should fail")
@@ -90,16 +91,30 @@ suite("txn_insert_inject_case", "nonConcurrent") {
             assertTrue(e.getMessage().contains("dbug_be_memtable_submit_flush_error"))
         }
 
-        disableDebugPoint()
-        sql """ insert into ${table}_0 select * from ${table}_1; """
+        disableDebugPoint("FlushToken.submit_flush_error")
         sql """ commit"""
     } catch (Exception e) {
         logger.error("failed", e)
     } finally {
         sql """ rollback """
-        disableDebugPoint()
+        disableDebugPoint("FlushToken.submit_flush_error")
         GetDebugPoint().disableDebugPointForAllBEs("FlushToken.submit_flush_error")
     }
+
+    try {
+        enableDebugPoint("FragmentMgr.exec_plan_fragment.failed")
+        sql """ begin """
+        test {
+            sql """ insert into ${table}_0 select * from ${table}_1; """
+            exception "FragmentMgr.exec_plan_fragment.failed"
+        }
+        disableDebugPoint("FragmentMgr.exec_plan_fragment.failed")
+        sql """ insert into ${table}_0 select * from ${table}_1; """
+        sql """ commit """
+    } finally {
+        sql """ rollback """
+        disableDebugPoint("FragmentMgr.exec_plan_fragment.failed")
+    }
     sql "sync"
     order_qt_select1 """select * from ${table}_0"""