Skip to content

Commit 82e90ef

Browse files
sollhuisollhui
authored
[feat](job) show first error message for routine load (#65553)
### What problem does this PR solve? PR #55666 added the first data quality error to load responses, but Routine Load was not covered. Routine Load transaction attachments only propagated the error log URL, so users still had to open the URL to inspect the actual error. ### Design This PR: - Captures the first data quality error for both successful and failed Routine Load tasks. - Propagates `first_error_msg` through the Routine Load Thrift transaction attachment and cloud transaction attachment. - Limits the message using `first_error_msg_max_length` when constructing the FE transaction attachment. - Forwards Routine Load attachments through the live cloud abort transaction path. - Adds a separate `FirstErrorMsg` column to `SHOW ROUTINE LOAD`. - Appends `FIRST_ERROR_MSG` to `information_schema.routine_load_jobs` without changing existing column ordinals. - Keeps `ERROR_LOG_URLS` unchanged instead of embedding the message in the URL field. - Treats Error URLs and the first error message as transient diagnostics: - They are not persisted in Routine Load job metadata or cloud progress. - They are not restored during transaction replay. - They are cleared together when the current error-statistics window is reset. - Keeps the behavior of successful ordinary Stream Load unchanged. ### Example `first_error_msg` output: ```text first_error_msg: no partition for this tuple. Src line: 100|bad_row ``` The first error can be queried directly: ``` SHOW ROUTINE LOAD FOR job_name; ``` or ``` SELECT ERROR_LOG_URLS, FIRST_ERROR_MSG FROM information_schema.routine_load_jobs WHERE JOB_NAME = 'job_name'; ```
1 parent f59397e commit 82e90ef

15 files changed

Lines changed: 282 additions & 13 deletions

File tree

‎be/src/information_schema/schema_routine_load_job_scanner.cpp‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ std::vector<SchemaScanner::ColumnDesc> SchemaRoutineLoadJobScanner::_s_tbls_colu
5555
{"CURRENT_ABORT_TASK_NUM", TYPE_INT, sizeof(int32_t), true},
5656
{"IS_ABNORMAL_PAUSE", TYPE_BOOLEAN, sizeof(int8_t), true},
5757
{"COMPUTE_GROUP", TYPE_STRING, sizeof(StringRef), true},
58+
{"FIRST_ERROR_MSG", TYPE_STRING, sizeof(StringRef), true},
5859
};
5960

6061
SchemaRoutineLoadJobScanner::SchemaRoutineLoadJobScanner()
@@ -175,6 +176,9 @@ Status SchemaRoutineLoadJobScanner::_fill_block_impl(Block* block) {
175176
case 20: // COMPUTE_GROUP
176177
column_value = job_info.__isset.compute_group ? job_info.compute_group : "";
177178
break;
179+
case 21: // FIRST_ERROR_MSG
180+
column_value = job_info.__isset.first_error_msg ? job_info.first_error_msg : "";
181+
break;
178182
}
179183

180184
str_refs[row_idx] =

‎be/src/load/stream_load/stream_load_executor.cpp‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,9 @@ Status StreamLoadExecutor::execute_plan_fragment(
110110
*status = Status::DataQualityError("too many filtered rows");
111111
}
112112
}
113+
if (ctx->load_type == TLoadType::ROUTINE_LOAD || !status->ok()) {
114+
ctx->first_error_msg = state->get_first_error_msg();
115+
}
113116

114117
if (status->ok()) {
115118
DorisMetrics::instance()->stream_receive_bytes_total->increment(ctx->receive_bytes);
@@ -118,7 +121,6 @@ Status StreamLoadExecutor::execute_plan_fragment(
118121
LOG(WARNING) << "fragment execute failed"
119122
<< ", err_msg=" << status->to_string() << ", " << ctx->brief();
120123
ctx->number_loaded_rows = 0;
121-
ctx->first_error_msg = state->get_first_error_msg();
122124
// cancel body_sink, make sender known it
123125
if (ctx->body_sink != nullptr) {
124126
ctx->body_sink->cancel(status->to_string());
@@ -428,6 +430,9 @@ bool StreamLoadExecutor::collect_load_stat(StreamLoadContext* ctx, TTxnCommitAtt
428430
rl_attach.__set_receivedBytes(ctx->receive_bytes);
429431
rl_attach.__set_loadedBytes(ctx->loaded_bytes);
430432
rl_attach.__set_loadCostMs(ctx->load_cost_millis);
433+
if (!ctx->first_error_msg.empty()) {
434+
rl_attach.__set_firstErrorMsg(ctx->first_error_msg);
435+
}
431436

432437
attach->rlTaskTxnCommitAttachment = rl_attach;
433438
attach->__isset.rlTaskTxnCommitAttachment = true;

‎fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -706,6 +706,7 @@ public class SchemaTable extends Table {
706706
.column("CURRENT_ABORT_TASK_NUM", ScalarType.createType(PrimitiveType.INT))
707707
.column("IS_ABNORMAL_PAUSE", ScalarType.createType(PrimitiveType.BOOLEAN))
708708
.column("COMPUTE_GROUP", ScalarType.createStringType())
709+
.column("FIRST_ERROR_MSG", ScalarType.createStringType())
709710
.build())
710711
)
711712
.put("load_jobs",

‎fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1841,7 +1841,7 @@ public void abortTransaction(Long dbId, Long transactionId, String reason,
18411841

18421842
AbortTxnResponse abortTxnResponse = null;
18431843
try {
1844-
abortTxnResponse = abortTransactionImpl(dbId, transactionId, reason, null);
1844+
abortTxnResponse = abortTransactionImpl(dbId, transactionId, reason, txnCommitAttachment);
18451845
} finally {
18461846
handleAfterAbort(abortTxnResponse, txnCommitAttachment, transactionId,
18471847
callbackInfo.first, callbackInfo.second);
@@ -1905,6 +1905,10 @@ private AbortTxnResponse abortTransactionImpl(Long dbId, Long transactionId, Str
19051905
builder.setTxnId(transactionId);
19061906
builder.setReason(reason);
19071907
builder.setCloudUniqueId(Config.cloud_unique_id);
1908+
if (txnCommitAttachment instanceof RLTaskTxnCommitAttachment) {
1909+
builder.setCommitAttachment(TxnUtil.rlTaskTxnCommitAttachmentToPb(
1910+
(RLTaskTxnCommitAttachment) txnCommitAttachment));
1911+
}
19081912

19091913
final AbortTxnRequest abortTxnRequest = builder.build();
19101914
AbortTxnResponse abortTxnResponse = null;

‎fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/TxnUtil.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -256,6 +256,9 @@ public static TxnCommitAttachmentPB rlTaskTxnCommitAttachmentToPb(RLTaskTxnCommi
256256
if (rtTaskTxnCommitAttachment.getErrorLogUrl() != null) {
257257
builder.setErrorLogUrl(rtTaskTxnCommitAttachment.getErrorLogUrl());
258258
}
259+
if (rtTaskTxnCommitAttachment.getFirstErrorMsg() != null) {
260+
builder.setFirstErrorMsg(rtTaskTxnCommitAttachment.getFirstErrorMsg());
261+
}
259262

260263
attachementBuilder.setRlTaskTxnCommitAttachment(builder.build());
261264
return attachementBuilder.build();

‎fe/fe-core/src/main/java/org/apache/doris/load/routineload/RLTaskTxnCommitAttachment.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
package org.apache.doris.load.routineload;
1919

2020
import org.apache.doris.cloud.proto.Cloud.RLTaskTxnCommitAttachmentPB;
21+
import org.apache.doris.common.Config;
2122
import org.apache.doris.load.routineload.kafka.KafkaProgress;
2223
import org.apache.doris.load.routineload.kinesis.KinesisProgress;
2324
import org.apache.doris.thrift.TRLTaskTxnCommitAttachment;
@@ -26,6 +27,7 @@
2627
import org.apache.doris.transaction.TxnCommitAttachment;
2728

2829
import com.google.gson.annotations.SerializedName;
30+
import org.apache.commons.lang3.StringUtils;
2931

3032
// {"progress": "", "backendId": "", "taskSignature": "", "numOfErrorData": "",
3133
// "numOfTotalData": "", "taskId": "", "jobId": ""}
@@ -46,6 +48,7 @@ public class RLTaskTxnCommitAttachment extends TxnCommitAttachment {
4648
@SerializedName(value = "pro")
4749
private RoutineLoadProgress progress;
4850
private String errorLogUrl;
51+
private String firstErrorMsg;
4952

5053
public RLTaskTxnCommitAttachment() {
5154
super(TransactionState.LoadJobSourceType.ROUTINE_LOAD_TASK);
@@ -75,6 +78,9 @@ public RLTaskTxnCommitAttachment(TRLTaskTxnCommitAttachment rlTaskTxnCommitAttac
7578
if (rlTaskTxnCommitAttachment.isSetErrorLogUrl()) {
7679
this.errorLogUrl = rlTaskTxnCommitAttachment.getErrorLogUrl();
7780
}
81+
if (rlTaskTxnCommitAttachment.isSetFirstErrorMsg()) {
82+
this.firstErrorMsg = abbreviateFirstErrorMsg(rlTaskTxnCommitAttachment.getFirstErrorMsg());
83+
}
7884
}
7985

8086
public RLTaskTxnCommitAttachment(RLTaskTxnCommitAttachmentPB rlTaskTxnCommitAttachment) {
@@ -92,6 +98,11 @@ public RLTaskTxnCommitAttachment(RLTaskTxnCommitAttachmentPB rlTaskTxnCommitAtta
9298

9399
this.progress = progress;
94100
this.errorLogUrl = rlTaskTxnCommitAttachment.getErrorLogUrl();
101+
this.firstErrorMsg = abbreviateFirstErrorMsg(rlTaskTxnCommitAttachment.getFirstErrorMsg());
102+
}
103+
104+
private static String abbreviateFirstErrorMsg(String firstErrorMsg) {
105+
return StringUtils.abbreviate(firstErrorMsg, Config.first_error_msg_max_length);
95106
}
96107

97108
public long getJobId() {
@@ -134,6 +145,10 @@ public String getErrorLogUrl() {
134145
return errorLogUrl;
135146
}
136147

148+
public String getFirstErrorMsg() {
149+
return firstErrorMsg;
150+
}
151+
137152
@Override
138153
public String toString() {
139154
return "RLTaskTxnCommitAttachment [filteredRows=" + filteredRows

‎fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java‎

Lines changed: 24 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -274,8 +274,10 @@ public boolean isFinalState() {
274274
protected Expr deleteCondition;
275275
// TODO(ml): error sample
276276

277-
// save the latest 3 error log urls
278-
private Queue<String> errorLogUrls = EvictingQueue.create(3);
277+
// Save the latest 3 error log URLs in memory. The corresponding first error message
278+
// uses the same lifecycle and should not be persisted with the job.
279+
private transient Queue<String> errorLogUrls = EvictingQueue.create(3);
280+
private transient String firstErrorMsg = "";
279281

280282
@SerializedName("ccid")
281283
private String cloudClusterId;
@@ -809,6 +811,10 @@ public Queue<String> getErrorLogUrls() {
809811
return errorLogUrls;
810812
}
811813

814+
public String getFirstErrorMsg() {
815+
return Strings.nullToEmpty(firstErrorMsg);
816+
}
817+
812818
// RoutineLoadScheduler will run this method at fixed interval, and renew the timeout tasks
813819
public void processTimeoutTasks() {
814820
writeLock();
@@ -954,10 +960,7 @@ private void updateNumOfData(long numOfTotalRows, long numOfErrorRows, long unse
954960
+ "when current total rows is more than base or the filter ratio is more than the max")
955961
.build());
956962
}
957-
// reset currentTotalNum, currentErrorNum and otherMsg
958-
this.jobStatistic.currentErrorRows = 0;
959-
this.jobStatistic.currentTotalRows = 0;
960-
this.otherMsg = "";
963+
resetCurrentErrorStatistics();
961964
this.jobStatistic.currentAbortedTaskNum = 0;
962965
} else if (this.jobStatistic.currentErrorRows > maxErrorNum
963966
|| (this.jobStatistic.currentTotalRows > 0
@@ -977,13 +980,18 @@ private void updateNumOfData(long numOfTotalRows, long numOfErrorRows, long unse
977980
"current error rows is more than max_error_number "
978981
+ "or the max_filter_ratio is more than the value set"), isReplay);
979982
}
980-
// reset currentTotalNum, currentErrorNum and otherMsg
981-
this.jobStatistic.currentErrorRows = 0;
982-
this.jobStatistic.currentTotalRows = 0;
983-
this.otherMsg = "";
983+
resetCurrentErrorStatistics();
984984
}
985985
}
986986

987+
private void resetCurrentErrorStatistics() {
988+
this.jobStatistic.currentErrorRows = 0;
989+
this.jobStatistic.currentTotalRows = 0;
990+
this.otherMsg = "";
991+
this.errorLogUrls.clear();
992+
this.firstErrorMsg = "";
993+
}
994+
987995
protected void replayUpdateProgress(RLTaskTxnCommitAttachment attachment) {
988996
try {
989997
updateNumOfData(attachment.getTotalRows(), attachment.getFilteredRows(), attachment.getUnselectedRows(),
@@ -1408,8 +1416,10 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn
14081416
routineLoadTaskInfo.handleTaskByTxnCommitAttachment(rlTaskTxnCommitAttachment);
14091417
}
14101418

1411-
if (rlTaskTxnCommitAttachment != null && !Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getErrorLogUrl())) {
1419+
if (rlTaskTxnCommitAttachment != null
1420+
&& !Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getErrorLogUrl())) {
14121421
errorLogUrls.add(rlTaskTxnCommitAttachment.getErrorLogUrl());
1422+
firstErrorMsg = Strings.nullToEmpty(rlTaskTxnCommitAttachment.getFirstErrorMsg());
14131423
}
14141424

14151425
routineLoadTaskInfo.setTxnStatus(txnStatus);
@@ -1712,6 +1722,7 @@ public List<String> getShowInfo() {
17121722
row.add(userIdentity.getQualifiedUser());
17131723
row.add(comment);
17141724
row.add(getClusterInfo());
1725+
row.add(getFirstErrorMsg());
17151726
return row;
17161727
} finally {
17171728
readUnlock();
@@ -1951,6 +1962,8 @@ public void write(DataOutput out) throws IOException {
19511962

19521963
@Override
19531964
public void gsonPostProcess() throws IOException {
1965+
errorLogUrls = EvictingQueue.create(3);
1966+
firstErrorMsg = "";
19541967
if (tableId == 0) {
19551968
isMultiTable = true;
19561969
}

‎fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowRoutineLoadCommand.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,7 @@ public class ShowRoutineLoadCommand extends ShowCommand {
110110
.add("User")
111111
.add("Comment")
112112
.add("ComputeGroup")
113+
.add("FirstErrorMsg")
113114
.build();
114115

115116
private final LabelNameInfo labelNameInfo;

‎fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5582,6 +5582,7 @@ public TFetchRoutineLoadJobResult fetchRoutineLoadJob(TFetchRoutineLoadJobReques
55825582
jobInfo.setLag(job.getLag());
55835583
jobInfo.setReasonOfStateChanged(job.getStateReason());
55845584
jobInfo.setErrorLogUrls(Joiner.on(", ").join(job.getErrorLogUrls()));
5585+
jobInfo.setFirstErrorMsg(job.getFirstErrorMsg());
55855586
jobInfo.setUserName(job.getUserIdentity().getQualifiedUser());
55865587
jobInfo.setCurrentAbortTaskNum(job.getJobStatistic().currentAbortedTaskNum);
55875588
jobInfo.setIsAbnormalPause(job.isAbnormalPause());

‎fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,15 +36,18 @@
3636
import org.apache.doris.common.FeMetaVersion;
3737
import org.apache.doris.common.LabelAlreadyUsedException;
3838
import org.apache.doris.common.UserException;
39+
import org.apache.doris.load.routineload.RLTaskTxnCommitAttachment;
3940
import org.apache.doris.transaction.GlobalTransactionMgrIface;
4041
import org.apache.doris.transaction.TransactionState;
42+
import org.apache.doris.transaction.TxnStateChangeCallback;
4143

4244
import com.google.common.collect.Lists;
4345
import org.junit.After;
4446
import org.junit.Assert;
4547
import org.junit.Before;
4648
import org.junit.Test;
4749
import org.junit.jupiter.api.Assertions;
50+
import org.mockito.ArgumentCaptor;
4851
import org.mockito.MockedStatic;
4952
import org.mockito.Mockito;
5053

@@ -306,6 +309,53 @@ public void testAbortTransaction() throws Exception {
306309
}
307310
}
308311

312+
@Test
313+
public void testAbortRoutineLoadTransactionWithAttachment() throws Exception {
314+
long transactionId = 123534;
315+
long jobId = 1001;
316+
RLTaskTxnCommitAttachment attachment = new RLTaskTxnCommitAttachment(
317+
Cloud.RLTaskTxnCommitAttachmentPB.newBuilder()
318+
.setJobId(jobId)
319+
.setTaskId(Cloud.UniqueIdPB.newBuilder().setHi(1).setLo(2))
320+
.setProgress(Cloud.RoutineLoadProgressPB.newBuilder())
321+
.setFirstErrorMsg("invalid source row")
322+
.build());
323+
TxnStateChangeCallback callback = Mockito.mock(TxnStateChangeCallback.class);
324+
Mockito.when(callback.getId()).thenReturn(jobId);
325+
masterTransMgr.getCallbackFactory().addCallback(callback);
326+
327+
MetaServiceProxy mockProxy = Mockito.mock(MetaServiceProxy.class);
328+
try (MockedStatic<MetaServiceProxy> mockedStatic = Mockito.mockStatic(MetaServiceProxy.class)) {
329+
mockedStatic.when(MetaServiceProxy::getInstance).thenReturn(mockProxy);
330+
Mockito.doAnswer(invocation -> {
331+
Cloud.AbortTxnRequest request = invocation.getArgument(0);
332+
Assert.assertTrue(request.hasCommitAttachment());
333+
Assert.assertEquals("invalid source row", request.getCommitAttachment()
334+
.getRlTaskTxnCommitAttachment().getFirstErrorMsg());
335+
return AbortTxnResponse.newBuilder()
336+
.setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
337+
.setCode(MetaServiceCode.OK).setMsg("OK"))
338+
.setTxnInfo(buildTxnInfo(transactionId).toBuilder()
339+
.setStatus(Cloud.TxnStatusPB.TXN_STATUS_ABORTED)
340+
.setReason("data quality error")
341+
.setCommitAttachment(request.getCommitAttachment()))
342+
.build();
343+
}).when(mockProxy).abortTxn(Mockito.any());
344+
345+
masterTransMgr.abortTransaction(CatalogTestUtil.testDbId1, transactionId,
346+
"data quality error", attachment, Lists.newArrayList());
347+
348+
ArgumentCaptor<TransactionState> txnStateCaptor = ArgumentCaptor.forClass(TransactionState.class);
349+
Mockito.verify(callback).afterAborted(txnStateCaptor.capture(), Mockito.eq(true),
350+
Mockito.eq("data quality error"));
351+
RLTaskTxnCommitAttachment callbackAttachment =
352+
(RLTaskTxnCommitAttachment) txnStateCaptor.getValue().getTxnCommitAttachment();
353+
Assert.assertEquals("invalid source row", callbackAttachment.getFirstErrorMsg());
354+
} finally {
355+
masterTransMgr.getCallbackFactory().removeCallback(jobId);
356+
}
357+
}
358+
309359
@Test
310360
public void testAbortTransactionByLabel() throws Exception {
311361
MetaServiceProxy mockProxy = Mockito.mock(MetaServiceProxy.class);

0 commit comments

Comments
 (0)