Skip to content

[fix](fe) Return transaction insert execution errors #67386

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mymeiyi wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[fix](fe) Return transaction insert execution errors #67386

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2777,7 +2777,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;
Expand Down
17 changes: 1 addition & 16 deletions ...a/org/apache/doris/nereids/trees/plans/commands/insert/OlapGroupCommitInsertExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand Down Expand Up @@ -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
Expand Down
6 changes: 5 additions & 1 deletion ...rc/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -312,6 +312,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())) {
Expand All @@ -323,7 +328,6 @@ protected void onFail(Throwable t) {
}
String finalErrorMsg = InsertUtils.getFinalErrorMsg(errMsg, firstErrorMsgPart, urlPart);
ctx.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR, finalErrorMsg);
recordLoadJob(ctx.getCurrentUserIdentity());
}
Comment thread
mymeiyi marked this conversation as resolved.

private void recordLoadJob(UserIdentity userIdentity) {
Expand Down
1 change: 1 addition & 0 deletions ...main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapTxnInsertExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
19 changes: 1 addition & 18 deletions ...n/java/org/apache/doris/nereids/trees/plans/commands/insert/RemoteOlapInsertExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@

import org.apache.doris.catalog.Env;
import org.apache.doris.common.AuthenticationException;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.LabelAlreadyUsedException;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.QuotaExceedException;
Expand Down Expand Up @@ -54,7 +53,6 @@
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;

Expand Down Expand Up @@ -260,20 +258,6 @@ protected void abortTransactionOnFail() throws Exception {
}
}

private String buildFinalErrorMessage(Throwable t) {
String localErrMsg = t.getMessage() == null ? "unknown reason" : t.getMessage();
String firstErrorMsgPart = "";
String urlPart = "";
if (!Strings.isNullOrEmpty(coordinator.getFirstErrorMsg())) {
firstErrorMsgPart = StringUtils.abbreviate(coordinator.getFirstErrorMsg(),
org.apache.doris.common.Config.first_error_msg_max_length);
}
if (!Strings.isNullOrEmpty(coordinator.getTrackingUrl())) {
urlPart = coordinator.getTrackingUrl();
}
return InsertUtils.getFinalErrorMsg(localErrMsg, firstErrorMsgPart, urlPart);
}

@Override
protected void onFail(Throwable t) {
errMsg = t.getMessage() == null ? "unknown reason" : t.getMessage();
Expand All @@ -287,8 +271,7 @@ protected void onFail(Throwable t) {
labelName, queryId, txnId, abortTxnException);
}
}
String finalErrorMsg = buildFinalErrorMessage(t);
ctx.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR, finalErrorMsg);
setErrorState();
}

@Override
Expand Down
35 changes: 25 additions & 10 deletions regression-test/suites/insert_p0/transaction/txn_insert_inject_case.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ suite("txn_insert_inject_case", "nonConcurrent") {
"""
}
GetDebugPoint().disableDebugPointForAllBEs("FlushToken.submit_flush_error")
GetDebugPoint().disableDebugPointForAllBEs("FragmentMgr.exec_plan_fragment.failed")
Comment thread
mymeiyi marked this conversation as resolved.
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]) """

Expand All @@ -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; """
Expand All @@ -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")
Expand All @@ -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"""

Expand Down
Loading