This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/load-scanner-restart in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ecbf4150d72031399d7336eb66409cac09a603e3 Author: Caideyipi <[email protected]> AuthorDate: Tue Nov 3 10:47:45 2026 +0800 Fix automatic LOAD retries during region recovery --- .../plan/analyze/load/LoadTsFileAnalyzer.java | 11 +++ .../load/TreeSchemaAutoCreatorAndVerifier.java | 8 ++ .../load/LoadTsFileAnalyzerDiagnosticsTest.java | 92 +++++++++++++++++++++- 3 files changed, 110 insertions(+), 1 deletion(-) 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 802e4a1697e..57b912a57d1 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 @@ -22,6 +22,7 @@ package org.apache.iotdb.db.queryengine.plan.analyze.load; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.auth.AuthException; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.queryengine.common.SessionInfo; import org.apache.iotdb.commons.queryengine.common.SqlDialect; @@ -898,6 +899,16 @@ public class LoadTsFileAnalyzer implements AutoCloseable { private boolean setTemporaryUnavailableStatusIfNecessary( final IAnalysis analysis, final Throwable throwable) { + if (throwable instanceof IoTDBRuntimeException + && ((IoTDBRuntimeException) throwable).getErrorCode() + == TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode()) { + analysis.setFinishQueryAfterAnalyze(true); + analysis.setFailStatus( + RpcUtils.getStatus( + TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION, throwable.getMessage())); + setRealStatement(analysis); + return true; + } if (isTemporaryUnavailableDueToPipeSchemaNotReady(throwable)) { setFailAnalysisForTemporaryUnavailablePipeSchema(analysis, throwable); return true; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java index 37911092405..02db897727e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java @@ -27,6 +27,7 @@ import org.apache.iotdb.commons.client.IClientManager; import org.apache.iotdb.commons.client.exception.ClientManagerException; import org.apache.iotdb.commons.consensus.ConfigRegionId; import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.schema.SchemaConstant; @@ -305,6 +306,13 @@ public class TreeSchemaAutoCreatorAndVerifier { } private void handleException(Exception e, String statementString) throws SemanticException { + // Region groups may still be recovering after a restart. Preserve the status so the loader + // can distinguish that temporary unavailability from a permanent schema error. + if (e instanceof IoTDBRuntimeException + && ((IoTDBRuntimeException) e).getErrorCode() + == TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode()) { + throw (IoTDBRuntimeException) e; + } LOGGER.warn(DataNodeQueryMessages.AUTO_CREATE_OR_VERIFY_SCHEMA_ERROR, e); throw new SemanticException( String.format( 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 index 65eb3d27710..44c2d701307 100644 --- 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 @@ -21,6 +21,7 @@ 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.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadAnalyzeException; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; @@ -28,12 +29,16 @@ 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.analyze.schema.SchemaValidator; 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.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.write.schema.MeasurementSchema; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -48,17 +53,20 @@ import org.powermock.modules.junit4.PowerMockRunner; import org.powermock.reflect.Whitebox; import java.io.File; +import java.io.FileOutputStream; import java.io.IOException; import java.util.Collections; import java.util.Optional; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyList; 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) +@PrepareForTest({LoadTsFileAnalyzer.class, SchemaValidator.class}) public class LoadTsFileAnalyzerDiagnosticsTest { @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); @@ -82,6 +90,88 @@ public class LoadTsFileAnalyzerDiagnosticsTest { IoTDBDescriptor.getInstance().getConfig().setDataNodeId(originalDataNodeId); } + @Test + public void testUnavailableSchemaRegionsShouldRemainRetryable() throws Exception { + final TSStatus status = analyzeAfterSchemaFailure(TSStatusCode.NO_AVAILABLE_REGION_GROUP); + + Assert.assertEquals( + TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode(), status.getCode()); + Assert.assertTrue(status.getMessage().contains("schema request failed")); + } + + @Test + public void testPermanentSchemaErrorShouldNotBecomeRetryable() throws Exception { + final TSStatus status = analyzeAfterSchemaFailure(TSStatusCode.METADATA_ERROR); + + Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), status.getCode()); + Assert.assertTrue(status.getMessage().contains("schema request failed")); + } + + @Test + public void testUnavailableRegionsDuringSingleFileAnalysisShouldRetry() throws Exception { + final File file = temporaryFolder.newFile("regions-recovering.tsfile"); + try (final FileOutputStream output = new FileOutputStream(file)) { + output.write(1); + } + PowerMockito.whenNew(TsFileSequenceReader.class) + .withArguments(file.getAbsolutePath()) + .thenThrow( + new IoTDBRuntimeException( + "region request failed", TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode())); + + try (final LoadTsFileAnalyzer analyzer = + new LoadTsFileAnalyzer( + LoadTsFileStatement.createUnchecked(file.getPath()), + false, + new MPPQueryContext(new QueryId("load_region_recovery")))) { + final IAnalysis analysis = analyzer.analyzeFileByFile(new Analysis()); + + Assert.assertTrue(analysis.isFinishQueryAfterAnalyze()); + Assert.assertEquals( + analysis.getFailStatus().getMessage(), + TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode(), + analysis.getFailStatus().getCode()); + Assert.assertEquals("region request failed", analysis.getFailStatus().getMessage()); + Assert.assertTrue(file.exists()); + } + } + + private TSStatus analyzeAfterSchemaFailure(final TSStatusCode code) throws Exception { + PowerMockito.mockStatic(SchemaValidator.class); + when(SchemaValidator.validate( + any(), + anyList(), + anyList(), + anyList(), + anyList(), + anyList(), + anyList(), + anyBoolean(), + any())) + .thenThrow(new IoTDBRuntimeException("schema request failed", code.getStatusCode())); + final File file = temporaryFolder.newFile("schema-failure.tsfile"); + final LoadTsFileStatement statement = LoadTsFileStatement.createUnchecked(file.getPath()); + statement.setAutoCreateDatabase(false); + + try (final LoadTsFileAnalyzer analyzer = + new LoadTsFileAnalyzer( + statement, false, new MPPQueryContext(new QueryId("load_schema_recovery")))) { + // Model a batch whose metadata was accumulated before the final schema flush. + final TreeSchemaAutoCreatorAndVerifier verifier = + new TreeSchemaAutoCreatorAndVerifier(analyzer); + final LoadTsFileTreeSchemaCache cache = Whitebox.getInternalState(verifier, "schemaCache"); + final IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create("root.sg.d"); + cache.addTimeSeries(device, new MeasurementSchema("s", TSDataType.INT64)); + cache.addIsAlignedCache(device, false, true); + Whitebox.setInternalState(analyzer, "treeSchemaAutoCreatorAndVerifier", verifier); + + final IAnalysis analysis = analyzer.analyzeFileByFile(new Analysis()); + Assert.assertTrue(analysis.isFinishQueryAfterAnalyze()); + Assert.assertTrue(file.exists()); + return analysis.getFailStatus(); + } + } + @Test public void testModelReadFailureReportsCurrentError() throws Exception { final File file = temporaryFolder.newFile("unreadable.tsfile");
