You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/10/20 10:02:59 UTC
(iotdb) branch master updated: Preserve LOAD failure details in logs and client responses (#18935)
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 9904ca34008 Preserve LOAD failure details in logs and client responses (#18935)
9904ca34008 is described below
commit 9904ca34008c94db7fe0c5d25a36c4ddf677cb98
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Tue Oct 20 18:02:51 2026 +0800
Preserve LOAD failure details in logs and client responses (#18935)
* Preserve LOAD failure details in logs and client responses
* Declare the PowerMock reflection test dependency
---
iotdb-core/datanode/pom.xml | 5 +
.../iotdb/db/i18n/DataNodeQueryMessages.java | 17 ++
.../iotdb/db/i18n/DataNodeQueryMessages.java | 17 ++
.../plan/analyze/load/LoadTsFileAnalyzer.java | 56 +++---
.../plan/scheduler/load/LoadFallbackHandler.java | 56 +++---
.../scheduler/load/LoadTsFileDispatcherImpl.java | 25 ++-
.../load/LoadTsFileAnalyzerDiagnosticsTest.java | 206 +++++++++++++++++++++
.../scheduler/load/LoadFallbackHandlerTest.java | 154 +++++++++++++++
.../load/LoadTsFileDispatcherImplTest.java | 84 ++++++++-
9 files changed, 566 insertions(+), 54 deletions(-)
diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml
index 48bc0490611..16368a3a6ae 100644
--- a/iotdb-core/datanode/pom.xml
+++ b/iotdb-core/datanode/pom.xml
@@ -289,6 +289,11 @@
<artifactId>powermock-api-mockito2</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-reflect</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index c89034e8831..d6fcfc64eb1 100644
--- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -20,6 +20,23 @@
package org.apache.iotdb.db.i18n;
public final class DataNodeQueryMessages {
+ public static final String EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477 =
+ "Failed to convert TsFile %s to tablets. Status: %s";
+ public static final String EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD =
+ "Failed to convert TsFile %s to tablets: %s";
+ public static final String MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3 =
+ "Tablet conversion returned no status";
+ public static final String EXCEPTION_FAILED_TO_READ_TSFILE_MODEL_INFORMATION_BEFORE_TABLET_CONVERSION_FOR_STATEMENT_ARG_ARG_F05C852A =
+ "Failed to read TsFile model information before tablet conversion for statement %s: %s";
+ public static final String EXCEPTION_FAILED_TO_LOAD_TSFILES_AFTER_TABLET_CONVERSION_DETAILS_ARG_53647EC5 =
+ "Failed to load TsFiles after tablet conversion. Details: %s";
+ public static final String LOG_LOAD_REJECTED_UUID_ARG_REGION_ARG_ENDPOINT_ARG_STATUS_ARG_8ED0679E =
+ "LOAD rejected: uuid={}, region={}, endpoint={}, status={}";
+ public static final String MESSAGE_TARGET_NODE_REJECTED_LOAD_WITHOUT_A_STATUS_OR_A_MESSAGE_F4E4C645 =
+ "Target node rejected LOAD without a status or a message";
+ public static final String MESSAGE_FAILED_TO_DISPATCH_LOAD_ARG_FOR_REGION_ARG_TO_NODE_ARG_ARG_57EFEC3A =
+ "Failed to dispatch LOAD %s for region %s to node %s: %s";
+
public static final String MESSAGE_SCHEMA_REGION_IS_UNAVAILABLE_PLEASE_RETRY_LATER_D642510A =
"Schema region is unavailable. Please retry later.";
diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index a3307bd07cb..e06c0ded22c 100644
--- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -20,6 +20,23 @@
package org.apache.iotdb.db.i18n;
public final class DataNodeQueryMessages {
+ public static final String EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477 =
+ "将 TsFile %s 转换为 tablet 失败。状态：%s";
+ public static final String EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD =
+ "将 TsFile %s 转换为 tablet 失败：%s";
+ public static final String MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3 =
+ "tablet 转换未返回状态";
+ public static final String EXCEPTION_FAILED_TO_READ_TSFILE_MODEL_INFORMATION_BEFORE_TABLET_CONVERSION_FOR_STATEMENT_ARG_ARG_F05C852A =
+ "执行语句 %s 时，在转换为 tablet 前读取 TsFile 模型信息失败：%s";
+ public static final String EXCEPTION_FAILED_TO_LOAD_TSFILES_AFTER_TABLET_CONVERSION_DETAILS_ARG_53647EC5 =
+ "将 TsFile 转换为 tablet 后加载仍失败。详情：%s";
+ public static final String LOG_LOAD_REJECTED_UUID_ARG_REGION_ARG_ENDPOINT_ARG_STATUS_ARG_8ED0679E =
+ "LOAD 被拒绝：uuid={}，region={}，endpoint={}，status={}";
+ public static final String MESSAGE_TARGET_NODE_REJECTED_LOAD_WITHOUT_A_STATUS_OR_A_MESSAGE_F4E4C645 =
+ "目标节点拒绝 LOAD，且未返回状态或消息";
+ public static final String MESSAGE_FAILED_TO_DISPATCH_LOAD_ARG_FOR_REGION_ARG_TO_NODE_ARG_ARG_57EFEC3A =
+ "分发 LOAD %s（region %s）到节点 %s 失败：%s";
+
public static final String MESSAGE_SCHEMA_REGION_IS_UNAVAILABLE_PLEASE_RETRY_LATER_D642510A =
"SchemaRegion 不可用，请稍后重试。";
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
index 52ab6f5ebd2..268ea2a222e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
@@ -804,13 +804,15 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
try {
getFileModelInfoBeforeTabletConversion();
} catch (Exception e1) {
- LOGGER.warn(
- DataNodeQueryMessages
- .LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_STATEMENT_ARG_BECAUSE_FAILED_TO_READ_MODEL_INFO,
- isTableModelStatement ? loadTsFileTableStatement : loadTsFileTreeStatement,
- e1.getMessage());
+ final String message =
+ String.format(
+ DataNodeQueryMessages
+ .EXCEPTION_FAILED_TO_READ_TSFILE_MODEL_INFORMATION_BEFORE_TABLET_CONVERSION_FOR_STATEMENT_ARG_ARG_F05C852A,
+ isTableModelStatement ? loadTsFileTableStatement : loadTsFileTreeStatement,
+ e1);
+ LOGGER.warn(message, e1);
analysis.setFailStatus(
- new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(e.getMessage()));
+ new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(message));
analysis.setFinishQueryAfterAnalyze(true);
setRealStatement(analysis);
return;
@@ -822,6 +824,7 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
for (int i = 0; i < tsFiles.size(); i++) {
final long startTime = System.nanoTime();
+ final String filePath = tsFiles.get(i).getPath();
try {
final TSStatus status =
isTableModelTsFile.get(i)
@@ -846,30 +849,37 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
.orElse(null);
if (status == null) {
- LOGGER.warn(
- DataNodeQueryMessages
- .LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_STATEMENT_ARG_STATUS_IS_NULL,
- isTableModelStatement ? loadTsFileTableStatement : loadTsFileTreeStatement);
+ final String message =
+ String.format(
+ DataNodeQueryMessages
+ .EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD,
+ filePath,
+ DataNodeQueryMessages.MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3);
+ LOGGER.warn(message);
analysis.setFailStatus(
- new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode())
- .setMessage(e.getMessage()));
+ new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(message));
break;
} else if (!loadTsFileDataTypeConverter.isSuccessful(status)) {
- LOGGER.warn(
- DataNodeQueryMessages.LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_STATEMENT_ARG_STATUS_ARG,
- isTableModelStatement ? loadTsFileTableStatement : loadTsFileTreeStatement,
- status);
- analysis.setFailStatus(status);
+ final String message =
+ String.format(
+ DataNodeQueryMessages
+ .EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477,
+ filePath,
+ status);
+ LOGGER.warn(message);
+ analysis.setFailStatus(status.deepCopy().setMessage(message));
break;
}
} catch (final Exception e2) {
- LOGGER.warn(
- DataNodeQueryMessages
- .LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_STATEMENT_ARG_BECAUSE_EXCEPTION_ARG,
- isTableModelStatement ? loadTsFileTableStatement : loadTsFileTreeStatement,
- e2.getMessage());
+ final String message =
+ String.format(
+ DataNodeQueryMessages
+ .EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD,
+ filePath,
+ e2);
+ LOGGER.warn(message, e2);
analysis.setFailStatus(
- new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(e.getMessage()));
+ new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(message));
break;
} finally {
LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java
index 448b28a6111..27839d168b4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java
@@ -34,6 +34,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.FileNotFoundException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -113,6 +114,7 @@ public class LoadFallbackHandler {
private void convertAndRetry() {
final LoadTsFileDataTypeConverter converter =
new LoadTsFileDataTypeConverter(queryContext, isGeneratedByPipe);
+ final List<String> failureDetails = new ArrayList<>();
final Iterator<Integer> iterator = failedTsFileNodeIndexes.iterator();
while (iterator.hasNext()) {
@@ -141,21 +143,34 @@ public class LoadFallbackHandler {
.LOAD_SUCCESSFULLY_CONVERTED_TSFILE_ARG_INTO_TABLETS_AND_INSERTED,
filePath);
} else {
- LOGGER.warn(
- DataNodeQueryMessages.LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_TSFILE_ARG_STATUS_ARG,
- filePath,
- status);
+ final String message =
+ status == null
+ ? String.format(
+ DataNodeQueryMessages
+ .EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD,
+ filePath,
+ DataNodeQueryMessages.MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3)
+ : String.format(
+ DataNodeQueryMessages
+ .EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477,
+ filePath,
+ status);
+ failureDetails.add(message);
+ LOGGER.warn(message);
}
} catch (final Exception e) {
- LOGGER.warn(
- DataNodeQueryMessages.LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_TSFILE_ARG_EXCEPTION_ARG,
- filePath,
- e.getMessage(),
- e);
+ final String message =
+ String.format(
+ DataNodeQueryMessages
+ .EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD,
+ filePath,
+ e);
+ failureDetails.add(message);
+ LOGGER.warn(message, e);
}
}
- resolveFinalState();
+ resolveFinalState(failureDetails);
}
private TSStatus executeTableModelConversion(
@@ -185,25 +200,20 @@ public class LoadFallbackHandler {
return converter.convertForTreeModel(statement).orElse(null);
}
- private void resolveFinalState() {
+ private void resolveFinalState(final List<String> failureDetails) {
if (failedTsFileNodeIndexes.isEmpty()) {
LOGGER.info(DataNodeQueryMessages.LOAD_ALL_FAILED_TSFILES_ARE_CONVERTED_TO_TABLETS);
stateMachine.transitionToFinished();
return;
}
- final String remainingFailedFiles = getFailedFilePathsString();
- LOGGER.warn(
- DataNodeQueryMessages
- .LOG_LOAD_FAILED_TO_LOAD_SOME_TSFILES_BY_CONVERTING_THEM_INTO_TABLETS_FAILED_TSFILES_ARG_7D9DB9C3,
- remainingFailedFiles);
-
- stateMachine.transitionToFailed(
- new LoadFileException(
- String.format(
- DataNodeQueryMessages
- .LOG_LOAD_FAILED_TO_LOAD_SOME_TSFILES_BY_CONVERTING_THEM_INTO_TABLETS_FAILED_TSFILES_ARG_7D9DB9C3,
- remainingFailedFiles)));
+ final String message =
+ String.format(
+ DataNodeQueryMessages
+ .EXCEPTION_FAILED_TO_LOAD_TSFILES_AFTER_TABLET_CONVERSION_DETAILS_ARG_53647EC5,
+ String.join("; ", failureDetails));
+ LOGGER.warn(message);
+ stateMachine.transitionToFailed(new LoadFileException(message));
}
// -------------------------------------------------------------------------
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
index ee1a86d1044..63a567e0d2f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
@@ -260,19 +260,37 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher, AutoCl
for (final TTsFilePieceReq request : requests) {
final TLoadResp loadResp = client.sendTsFilePieceNode(request);
if (!loadResp.isAccepted()) {
+ final TSStatus failureStatus =
+ loadResp.isSetStatus()
+ ? loadResp.getStatus()
+ : RpcUtils.getStatus(
+ TSStatusCode.DISPATCH_ERROR,
+ loadResp.isSetMessage()
+ ? loadResp.getMessage()
+ : DataNodeQueryMessages
+ .MESSAGE_TARGET_NODE_REJECTED_LOAD_WITHOUT_A_STATUS_OR_A_MESSAGE_F4E4C645);
final String errorCode =
loadResp.isSetStatus()
? String.valueOf(loadResp.getStatus().getCode())
: UserDataTransferErrorCode.REMOTE_REJECTED.name();
recordTransferAttempt(endPoint, false, errorCode, null);
transferAttemptAudited = true;
- LOGGER.warn(loadResp.getMessage());
- throw new FragmentInstanceDispatchException(loadResp.getStatus());
+ LOGGER.warn(
+ DataNodeQueryMessages
+ .LOG_LOAD_REJECTED_UUID_ARG_REGION_ARG_ENDPOINT_ARG_STATUS_ARG_8ED0679E,
+ uuid,
+ consensusGroupId,
+ endPoint,
+ failureStatus);
+ throw new FragmentInstanceDispatchException(failureStatus);
}
}
recordTransferAttempt(endPoint, true, null, null);
transferAttemptAudited = true;
}
+ } catch (final FragmentInstanceDispatchException e) {
+ // A receiver rejection is already audited and carries its original status and retry hints.
+ throw e;
} catch (final Exception e) {
if (!transferAttemptAudited) {
recordTransferAttempt(endPoint, false, null, e);
@@ -282,8 +300,9 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher, AutoCl
final String message =
String.format(
DataNodeQueryMessages
- .MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_BECAUSE_OF_EXCEPTION_ARG_2D8A483D,
+ .MESSAGE_FAILED_TO_DISPATCH_LOAD_ARG_FOR_REGION_ARG_TO_NODE_ARG_ARG_57EFEC3A,
uuid,
+ consensusGroupId,
endPoint,
e);
LOGGER.warn(message, e);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java
new file mode 100644
index 00000000000..65eb3d27710
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java
@@ -0,0 +1,206 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.analyze.load;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.load.LoadAnalyzeException;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
+import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
+import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.file.metadata.TableSchema;
+import org.apache.tsfile.read.TsFileSequenceReader;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.reflect.Whitebox;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Optional;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"})
+@PrepareForTest(LoadTsFileAnalyzer.class)
+public class LoadTsFileAnalyzerDiagnosticsTest {
+
+ @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private LoadTsFileDataTypeConverter converter;
+ private int originalDataNodeId;
+
+ @Before
+ public void setUp() throws Exception {
+ originalDataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+ IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0);
+ converter = mock(LoadTsFileDataTypeConverter.class);
+ when(converter.isSuccessful(any())).thenCallRealMethod();
+ PowerMockito.whenNew(LoadTsFileDataTypeConverter.class)
+ .withAnyArguments()
+ .thenReturn(converter);
+ }
+
+ @After
+ public void tearDown() {
+ IoTDBDescriptor.getInstance().getConfig().setDataNodeId(originalDataNodeId);
+ }
+
+ @Test
+ public void testModelReadFailureReportsCurrentError() throws Exception {
+ final File file = temporaryFolder.newFile("unreadable.tsfile");
+ PowerMockito.whenNew(TsFileSequenceReader.class)
+ .withArguments(file.getAbsolutePath(), true)
+ .thenThrow(new IOException("cannot read TsFile footer"));
+
+ final TSStatus status = convertAfterAnalysisFailure(file);
+
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), status.getCode());
+ Assert.assertTrue(status.getMessage().contains(file.getPath()));
+ Assert.assertTrue(status.getMessage().contains("cannot read TsFile footer"));
+ Assert.assertFalse(status.getMessage().contains("original analysis failure"));
+ }
+
+ @Test
+ public void testConversionExceptionReportsCurrentErrorForBothModels() throws Exception {
+ when(converter.convertForTreeModel(any()))
+ .thenThrow(new IllegalStateException("tree conversion failed"));
+ when(converter.convertForTableModel(any()))
+ .thenThrow(new IllegalStateException("table conversion failed"));
+
+ for (final boolean tableModel : new boolean[] {false, true}) {
+ mockModelReader(tableModel);
+ final File file = temporaryFolder.newFile(tableModel ? "table.tsfile" : "tree.tsfile");
+
+ final TSStatus status = convertAfterAnalysisFailure(file);
+
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), status.getCode());
+ Assert.assertTrue(status.getMessage().contains(file.getPath()));
+ Assert.assertTrue(
+ status
+ .getMessage()
+ .contains(tableModel ? "table conversion failed" : "tree conversion failed"));
+ Assert.assertFalse(status.getMessage().contains("original analysis failure"));
+ }
+ }
+
+ @Test
+ public void testConversionExceptionWithoutMessageStillHasDetails() throws Exception {
+ mockModelReader(false);
+ when(converter.convertForTreeModel(any())).thenThrow(new IllegalStateException());
+
+ final File file = temporaryFolder.newFile("no-message.tsfile");
+ final TSStatus status = convertAfterAnalysisFailure(file);
+
+ Assert.assertTrue(status.getMessage().contains(file.getPath()));
+ Assert.assertTrue(status.getMessage().contains(IllegalStateException.class.getName()));
+ }
+
+ @Test
+ public void testMissingConversionStatusHasExplicitReason() throws Exception {
+ mockModelReader(false);
+ when(converter.convertForTreeModel(any())).thenReturn(Optional.empty());
+
+ final File file = temporaryFolder.newFile("no-status.tsfile");
+ final TSStatus status = convertAfterAnalysisFailure(file);
+
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), status.getCode());
+ Assert.assertTrue(status.getMessage().contains(file.getPath()));
+ Assert.assertTrue(
+ status
+ .getMessage()
+ .contains(DataNodeQueryMessages.MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3));
+ Assert.assertFalse(status.getMessage().contains("original analysis failure"));
+ }
+
+ @Test
+ public void testConversionStatusKeepsCodeAndMetadataWithoutMutatingOriginal() throws Exception {
+ mockModelReader(false);
+ final TSStatus original =
+ new TSStatus(TSStatusCode.MULTIPLE_ERROR.getStatusCode())
+ .setMessage("tablet insertion failed")
+ .setSubStatus(
+ Collections.singletonList(
+ new TSStatus(TSStatusCode.NO_PERMISSION.getStatusCode())
+ .setMessage("missing WRITE_DATA privilege")))
+ .setRedirectNode(new TEndPoint("127.0.0.2", 10730))
+ .setNeedRetry(true);
+ final TSStatus snapshot = original.deepCopy();
+ when(converter.convertForTreeModel(any())).thenReturn(Optional.of(original));
+
+ final File file = temporaryFolder.newFile("denied.tsfile");
+ final TSStatus status = convertAfterAnalysisFailure(file);
+
+ Assert.assertNotSame(original, status);
+ Assert.assertEquals(snapshot, original);
+ Assert.assertEquals(original.getCode(), status.getCode());
+ Assert.assertEquals(original.getSubStatus(), status.getSubStatus());
+ Assert.assertEquals(original.getRedirectNode(), status.getRedirectNode());
+ Assert.assertEquals(original.isNeedRetry(), status.isNeedRetry());
+ Assert.assertTrue(status.getMessage().contains(file.getPath()));
+ Assert.assertTrue(status.getMessage().contains("missing WRITE_DATA privilege"));
+ }
+
+ private void mockModelReader(final boolean tableModel) throws Exception {
+ final TsFileSequenceReader reader = mock(TsFileSequenceReader.class);
+ when(reader.getTableSchemaMap())
+ .thenReturn(
+ tableModel
+ ? Collections.singletonMap("table1", mock(TableSchema.class))
+ : Collections.emptyMap());
+ PowerMockito.whenNew(TsFileSequenceReader.class).withAnyArguments().thenReturn(reader);
+ }
+
+ private TSStatus convertAfterAnalysisFailure(final File file) throws Exception {
+ try (final LoadTsFileAnalyzer analyzer =
+ new LoadTsFileAnalyzer(
+ LoadTsFileStatement.createUnchecked(file.getPath()).setConvertOnTypeMismatch(true),
+ false,
+ new MPPQueryContext(new QueryId("load_diagnostics")))) {
+ final Analysis analysis = new Analysis();
+ Whitebox.invokeMethod(
+ analyzer,
+ "executeTabletConversionOnException",
+ (IAnalysis) analysis,
+ new LoadAnalyzeException("original analysis failure"));
+ Assert.assertTrue(analysis.isFinishQueryAfterAnalyze());
+ return analysis.getFailStatus();
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandlerTest.java
new file mode 100644
index 00000000000..b765405277a
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandlerTest.java
@@ -0,0 +1,154 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.scheduler.load;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.execution.QueryState;
+import org.apache.iotdb.db.queryengine.execution.QueryStateMachine;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadSingleTsFileNode;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter;
+import org.apache.iotdb.rpc.RpcUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import com.google.common.util.concurrent.MoreExecutors;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"})
+@PrepareForTest(LoadFallbackHandler.class)
+public class LoadFallbackHandlerTest {
+
+ @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private LoadTsFileDataTypeConverter converter;
+
+ @Before
+ public void setUp() throws Exception {
+ // Progress-log formatting is unrelated to conversion results. Avoid instrumenting JDK Stream
+ // internals through this helper when running PowerMock on Java 17.
+ PowerMockito.suppress(
+ PowerMockito.method(LoadFallbackHandler.class, "getFailedFilePathsString"));
+ converter = mock(LoadTsFileDataTypeConverter.class);
+ when(converter.isSuccessful(any())).thenCallRealMethod();
+ PowerMockito.whenNew(LoadTsFileDataTypeConverter.class)
+ .withAnyArguments()
+ .thenReturn(converter);
+ }
+
+ @Test
+ public void testPartialFailureReportsEachFilesReason() throws Exception {
+ final List<LoadSingleTsFileNode> nodes =
+ Arrays.asList(
+ newNode("success.tsfile", false),
+ newNode("denied.tsfile", true),
+ newNode("exception.tsfile", false),
+ newNode("no-status.tsfile", false));
+ final TSStatus denied =
+ new TSStatus(TSStatusCode.MULTIPLE_ERROR.getStatusCode())
+ .setSubStatus(
+ Collections.singletonList(
+ new TSStatus(TSStatusCode.NO_PERMISSION.getStatusCode())
+ .setMessage("missing WRITE_DATA privilege")));
+ when(converter.convertForTableModel(any())).thenReturn(Optional.of(denied));
+ when(converter.convertForTreeModel(any()))
+ .thenReturn(Optional.of(RpcUtils.SUCCESS_STATUS))
+ .thenThrow(new IllegalStateException())
+ .thenReturn(Optional.empty());
+ final List<Integer> failedIndexes = new ArrayList<>(Arrays.asList(0, 1, 2, 3));
+ final QueryStateMachine stateMachine = newStateMachine();
+
+ new LoadFallbackHandler(mock(MPPQueryContext.class), false, nodes, failedIndexes, stateMachine)
+ .convertFailedTsFilesToTablets();
+
+ Assert.assertEquals(QueryState.FAILED, stateMachine.getState());
+ Assert.assertEquals(Arrays.asList(1, 2, 3), failedIndexes);
+ final TSStatus failure = stateMachine.getFailureStatus();
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), failure.getCode());
+ final String message = failure.getMessage();
+ Assert.assertFalse(message.contains("success.tsfile"));
+ for (int i = 1; i < nodes.size(); i++) {
+ Assert.assertTrue(message.contains(nodes.get(i).getTsFileResource().getTsFilePath()));
+ }
+ Assert.assertTrue(message.contains("missing WRITE_DATA privilege"));
+ Assert.assertTrue(message.contains(String.valueOf(TSStatusCode.NO_PERMISSION.getStatusCode())));
+ Assert.assertTrue(message.contains(IllegalStateException.class.getName()));
+ Assert.assertTrue(
+ message.contains(
+ DataNodeQueryMessages.MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3));
+ }
+
+ @Test
+ public void testSuccessfulConversionsStillFinishLoad() throws Exception {
+ final List<LoadSingleTsFileNode> nodes =
+ Arrays.asList(newNode("tree.tsfile", false), newNode("table.tsfile", true));
+ when(converter.convertForTreeModel(any())).thenReturn(Optional.of(RpcUtils.SUCCESS_STATUS));
+ when(converter.convertForTableModel(any())).thenReturn(Optional.of(RpcUtils.SUCCESS_STATUS));
+ final List<Integer> failedIndexes = new ArrayList<>(Arrays.asList(0, 1));
+ final QueryStateMachine stateMachine = newStateMachine();
+
+ new LoadFallbackHandler(mock(MPPQueryContext.class), false, nodes, failedIndexes, stateMachine)
+ .convertFailedTsFilesToTablets();
+
+ Assert.assertEquals(QueryState.FINISHED, stateMachine.getState());
+ Assert.assertTrue(failedIndexes.isEmpty());
+ }
+
+ private LoadSingleTsFileNode newNode(final String name, final boolean tableModel)
+ throws Exception {
+ final File file = temporaryFolder.newFile(name);
+ final LoadSingleTsFileNode node = mock(LoadSingleTsFileNode.class);
+ when(node.getTsFileResource()).thenReturn(new TsFileResource(file));
+ when(node.isTableModel()).thenReturn(tableModel);
+ when(node.getDatabase()).thenReturn(tableModel ? "test" : null);
+ return node;
+ }
+
+ private QueryStateMachine newStateMachine() {
+ final QueryStateMachine stateMachine =
+ new QueryStateMachine(
+ new QueryId("load_fallback"), MoreExecutors.newDirectExecutorService());
+ stateMachine.transitionToRunning();
+ return stateMachine;
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImplTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImplTest.java
index 18c09cf80e2..e553391307d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImplTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImplTest.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.consensus.DataRegionId;
@@ -31,16 +32,19 @@ import org.apache.iotdb.commons.partition.StorageExecutor;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.db.queryengine.plan.planner.plan.PlanFragment;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode;
+import org.apache.iotdb.db.queryengine.plan.scheduler.FragInstanceDispatchResult;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService;
import org.apache.iotdb.mpp.rpc.thrift.TLoadResp;
import org.apache.iotdb.mpp.rpc.thrift.TTsFilePieceReq;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TElasticFramedTransport;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.TException;
@@ -246,15 +250,85 @@ public class LoadTsFileDispatcherImplTest {
new LoadTsFilePieceNode(new PlanNodeId("piece"), new File("test.tsfile"));
try (LoadTsFileDispatcherImpl dispatcher = new LoadTsFileDispatcherImpl(clientManager, false)) {
dispatcher.setUuid("test-uuid");
- Assert.assertFalse(
- dispatcher
- .dispatch(null, Collections.singletonList(createFragmentInstance(pieceNode)))
- .get(10, TimeUnit.SECONDS)
- .isSuccessful());
+ final FragmentInstance instance = createFragmentInstance(pieceNode);
+ final FragInstanceDispatchResult result =
+ dispatcher.dispatch(null, Collections.singletonList(instance)).get(10, TimeUnit.SECONDS);
+ Assert.assertFalse(result.isSuccessful());
+ Assert.assertEquals(
+ TSStatusCode.DISPATCH_ERROR.getStatusCode(), result.getFailureStatus().getCode());
+ Assert.assertTrue(result.getFailureStatus().getMessage().contains("test-uuid"));
+ Assert.assertTrue(
+ result
+ .getFailureStatus()
+ .getMessage()
+ .contains(instance.getRegionReplicaSet().getRegionId().toString()));
+ Assert.assertTrue(result.getFailureStatus().getMessage().contains("127.0.0.1"));
}
Mockito.verify(client, Mockito.never()).sendTsFilePieceNode(Mockito.any());
}
+ @Test
+ public void testRemoteRejectionPreservesOriginalStatus() throws Exception {
+ final TSStatus status =
+ new TSStatus(TSStatusCode.MULTIPLE_ERROR.getStatusCode())
+ .setMessage("tablet insertion rejected")
+ .setSubStatus(
+ Collections.singletonList(
+ new TSStatus(TSStatusCode.NO_PERMISSION.getStatusCode())
+ .setMessage("missing WRITE_DATA privilege")))
+ .setRedirectNode(new TEndPoint("127.0.0.2", 10730))
+ .setNeedRetry(true);
+
+ final FragInstanceDispatchResult result =
+ dispatchWithResponse(new TLoadResp(false).setStatus(status));
+
+ Assert.assertFalse(result.isSuccessful());
+ Assert.assertEquals(status, result.getFailureStatus());
+ }
+
+ @Test
+ public void testRemoteRejectionWithoutStatusPreservesMessage() throws Exception {
+ final FragInstanceDispatchResult result =
+ dispatchWithResponse(new TLoadResp(false).setMessage("receiver disk is full"));
+
+ Assert.assertFalse(result.isSuccessful());
+ Assert.assertEquals(
+ TSStatusCode.DISPATCH_ERROR.getStatusCode(), result.getFailureStatus().getCode());
+ Assert.assertEquals("receiver disk is full", result.getFailureStatus().getMessage());
+ }
+
+ @Test
+ public void testRemoteRejectionWithoutDetailsHasExplicitReason() throws Exception {
+ final FragInstanceDispatchResult result = dispatchWithResponse(new TLoadResp(false));
+
+ Assert.assertFalse(result.isSuccessful());
+ Assert.assertEquals(
+ TSStatusCode.DISPATCH_ERROR.getStatusCode(), result.getFailureStatus().getCode());
+ Assert.assertEquals(
+ DataNodeQueryMessages
+ .MESSAGE_TARGET_NODE_REJECTED_LOAD_WITHOUT_A_STATUS_OR_A_MESSAGE_F4E4C645,
+ result.getFailureStatus().getMessage());
+ }
+
+ private static FragInstanceDispatchResult dispatchWithResponse(final TLoadResp response)
+ throws Exception {
+ final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> clientManager =
+ Mockito.mock(IClientManager.class);
+ final SyncDataNodeInternalServiceClient client =
+ Mockito.mock(SyncDataNodeInternalServiceClient.class);
+ Mockito.when(clientManager.borrowClient(Mockito.any())).thenReturn(client);
+ Mockito.when(client.getThriftMaxFrameSize()).thenReturn(4096);
+ Mockito.when(client.sendTsFilePieceNode(Mockito.any())).thenReturn(response);
+ final LoadTsFilePieceNode pieceNode =
+ new LoadTsFilePieceNode(new PlanNodeId("piece"), new File("test.tsfile"));
+ try (LoadTsFileDispatcherImpl dispatcher = new LoadTsFileDispatcherImpl(clientManager, false)) {
+ dispatcher.setUuid("test-uuid");
+ return dispatcher
+ .dispatch(null, Collections.singletonList(createFragmentInstance(pieceNode)))
+ .get(10, TimeUnit.SECONDS);
+ }
+ }
+
private static void assertFitsBothTransports(
final TTsFilePieceReq request, final int senderMaxFrameSize, final int receiverMaxFrameSize)
throws Exception {