From f685a12b3bde84b318d7eb0bf2267935100ac85b Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 4 Nov 2026 14:44:34 +0800 Subject: [PATCH 1/4] Fix automatic LOAD retries during region recovery (#19098) * Fix automatic LOAD retries during region recovery * Preserve wrapped region recovery errors during LOAD * Clarify region-unavailable LOAD retry semantics (cherry picked from commit 3ffe2958758ac2a7d9c372cc5e282d264e818eb2) --- .../plan/analyze/load/LoadTsFileAnalyzer.java | 37 +++++ .../LoadTsFileAnalyzerDiagnosticsTest.java | 139 +++++++++++++++++- 2 files changed, 175 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 4cdf80d5416c6..0758ebc0f57d7 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 @@ -27,6 +27,8 @@ import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.consensus.ConfigRegionId; import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.exception.IoTDBException; +import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.path.PatternTreeMap; import org.apache.iotdb.commons.schema.SchemaConstant; @@ -591,6 +593,18 @@ private Analysis executeTabletConversionOnException( private boolean setTemporaryUnavailableStatusIfNecessary( final Analysis analysis, final Throwable throwable) { + // Region unavailability can persist beyond recovery. Retain the file for retry; permanent + // region loss still requires operator intervention. + 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())); + analysis.setStatement(loadTsFileStatement); + return true; + } if (isTemporaryUnavailableDueToPipeSchemaNotReady(throwable)) { setFailAnalysisForTemporaryUnavailablePipeSchema(analysis, throwable); return true; @@ -849,12 +863,14 @@ private void doAutoCreateAndVerify() if (isTemporaryUnavailableDueToPipeSchemaNotReady(e)) { throw e; } + rethrowIfUnavailableRegion(e); LOGGER.warn("Auto create or verify schema error.", e); throw new SemanticException( String.format( "Auto create or verify schema error when executing statement %s. Detail: %s.", loadTsFileStatement, e.getMessage())); } catch (Exception e) { + rethrowIfUnavailableRegion(e); if (e.getCause() instanceof LoadAnalyzeTypeMismatchException && isConvertOnTypeMismatch) { throw (LoadAnalyzeTypeMismatchException) e.getCause(); } @@ -866,6 +882,27 @@ private void doAutoCreateAndVerify() } } + private void rethrowIfUnavailableRegion(final Exception e) { + // Preserve retryable region errors through the analyzer's schema-error wrapper. + Throwable current = e; + while (current != null) { + if (current instanceof IoTDBRuntimeException + && ((IoTDBRuntimeException) current).getErrorCode() + == TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode()) { + throw (IoTDBRuntimeException) current; + } + if (current instanceof IoTDBException + && ((IoTDBException) current).getErrorCode() + == TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode()) { + throw new IoTDBRuntimeException( + current.getMessage(), + current, + TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode()); + } + current = current.getCause(); + } + } + private void makeSureNoDuplicatedMeasurementsInDevices() throws LoadAnalyzeException { for (final Map.Entry> entry : schemaCache.getDevice2TimeSeries().entrySet()) { 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 816b79bd0134c..bceb77c89eef9 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,17 +21,25 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.exception.IoTDBException; +import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.db.exception.LoadAnalyzeException; +import org.apache.iotdb.db.exception.sql.SemanticException; 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.IPartitionFetcher; import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; +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.read.TsFileSequenceReader; +import org.apache.tsfile.write.schema.MeasurementSchema; import org.junit.Assert; import org.junit.Before; import org.junit.Rule; @@ -43,18 +51,22 @@ 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.FileOutputStream; import java.lang.reflect.Method; 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.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(); @@ -126,6 +138,131 @@ public void testConversionStatusKeepsMetadataWithoutMutatingOriginal() throws Ex Assert.assertTrue(status.getMessage().contains("missing WRITE_DATA privilege")); } + @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 testWrappedUnavailableSchemaRegionsShouldRemainRetryable() throws Exception { + final TSStatus status = + analyzeAfterSchemaFailure( + new SemanticException( + new IoTDBException( + "schema request failed", + TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode()))); + + Assert.assertEquals( + TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode(), status.getCode()); + Assert.assertTrue(status.getMessage().contains("schema request failed")); + } + + @Test + public void testWrappedPermanentSchemaErrorShouldNotBecomeRetryable() throws Exception { + final TSStatus status = + analyzeAfterSchemaFailure( + new SemanticException( + new IoTDBException( + "schema request failed", TSStatusCode.METADATA_ERROR.getStatusCode()))); + + 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()), + new MPPQueryContext(new QueryId("load_region_recovery")), + Mockito.mock(IPartitionFetcher.class), + Mockito.mock(ISchemaFetcher.class))) { + final Analysis analysis = analyzer.analyzeFileByFile(new Analysis()); + + Assert.assertTrue(analysis.isFinishQueryAfterAnalyze()); + Assert.assertEquals( + 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 { + return analyzeAfterSchemaFailure( + new IoTDBRuntimeException("schema request failed", code.getStatusCode())); + } + + private TSStatus analyzeAfterSchemaFailure(final RuntimeException exception) throws Exception { + PowerMockito.mockStatic(SchemaValidator.class); + when(SchemaValidator.validate( + any(), + anyList(), + anyList(), + anyList(), + anyList(), + anyList(), + anyList(), + anyBoolean(), + any())) + .thenThrow(exception); + + 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, + new MPPQueryContext(new QueryId("load_schema_recovery")), + Mockito.mock(IPartitionFetcher.class), + Mockito.mock(ISchemaFetcher.class))) { + final Object verifier = Whitebox.getInternalState(analyzer, "schemaAutoCreatorAndVerifier"); + final Object schemaCache = Whitebox.getInternalState(verifier, "schemaCache"); + final IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create("root.sg.d"); + final Method addTimeSeries = + schemaCache + .getClass() + .getDeclaredMethod("addTimeSeries", IDeviceID.class, MeasurementSchema.class); + addTimeSeries.setAccessible(true); + addTimeSeries.invoke(schemaCache, device, new MeasurementSchema("s", TSDataType.INT64)); + final Method addIsAlignedCache = + schemaCache + .getClass() + .getDeclaredMethod( + "addIsAlignedCache", IDeviceID.class, boolean.class, boolean.class); + addIsAlignedCache.setAccessible(true); + addIsAlignedCache.invoke(schemaCache, device, false, true); + + final Analysis analysis = analyzer.analyzeFileByFile(new Analysis()); + Assert.assertTrue(analysis.isFinishQueryAfterAnalyze()); + Assert.assertTrue(file.exists()); + return analysis.getFailStatus(); + } + } + private TSStatus convertAfterAnalysisFailure(final File file) throws Exception { try (final LoadTsFileAnalyzer analyzer = new LoadTsFileAnalyzer( From f089abde94f761ed0363ad95a43431236418e407 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 5 Nov 2026 18:10:27 +0800 Subject: [PATCH 2/4] Adapt LOAD recovery CP to 1.3 TsFile API --- .../plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 bceb77c89eef9..d3e7a39d698e8 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 @@ -38,6 +38,7 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.PlainDeviceID; import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.write.schema.MeasurementSchema; import org.junit.Assert; @@ -241,7 +242,7 @@ private TSStatus analyzeAfterSchemaFailure(final RuntimeException exception) thr Mockito.mock(ISchemaFetcher.class))) { final Object verifier = Whitebox.getInternalState(analyzer, "schemaAutoCreatorAndVerifier"); final Object schemaCache = Whitebox.getInternalState(verifier, "schemaCache"); - final IDeviceID device = IDeviceID.Factory.DEFAULT_FACTORY.create("root.sg.d"); + final IDeviceID device = new PlainDeviceID("root.sg.d"); final Method addTimeSeries = schemaCache .getClass() From d1f7b16175d087eecdc90d56bd0c70cd0baf6582 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 6 Nov 2026 09:41:27 +0800 Subject: [PATCH 3/4] Preserve LOAD conversion priority in 1.3 backport --- .../plan/analyze/load/LoadTsFileAnalyzer.java | 2 +- .../LoadTsFileAnalyzerDiagnosticsTest.java | 18 ++++++++++++++++++ 2 files changed, 19 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 0758ebc0f57d7..7dfd99ca62886 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 @@ -870,10 +870,10 @@ private void doAutoCreateAndVerify() "Auto create or verify schema error when executing statement %s. Detail: %s.", loadTsFileStatement, e.getMessage())); } catch (Exception e) { - rethrowIfUnavailableRegion(e); if (e.getCause() instanceof LoadAnalyzeTypeMismatchException && isConvertOnTypeMismatch) { throw (LoadAnalyzeTypeMismatchException) e.getCause(); } + rethrowIfUnavailableRegion(e); LOGGER.warn("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 d3e7a39d698e8..f3f71af8710ea 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 @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.exception.IoTDBException; import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.db.exception.LoadAnalyzeException; +import org.apache.iotdb.db.exception.LoadAnalyzeTypeMismatchException; import org.apache.iotdb.db.exception.sql.SemanticException; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; @@ -182,6 +183,23 @@ public void testWrappedPermanentSchemaErrorShouldNotBecomeRetryable() throws Exc Assert.assertTrue(status.getMessage().contains("schema request failed")); } + @Test + public void testWrappedTypeMismatchShouldKeepConversionPriority() throws Exception { + final LoadAnalyzeTypeMismatchException mismatch = + new LoadAnalyzeTypeMismatchException("schema type mismatch"); + mismatch.initCause( + new IoTDBRuntimeException( + "schema request failed", TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode())); + when(converter.convertForTreeModel(any())) + .thenReturn(Optional.of(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()))); + + final TSStatus status = analyzeAfterSchemaFailure(new SemanticException(mismatch)); + + // Successful conversion leaves the analysis failure status unset in 1.3. + Assert.assertNull(status); + Mockito.verify(converter).convertForTreeModel(any()); + } + @Test public void testUnavailableRegionsDuringSingleFileAnalysisShouldRetry() throws Exception { final File file = temporaryFolder.newFile("regions-recovering.tsfile"); From e79447202c8493067e5e0b27e65ce8514dcccd02 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 6 Nov 2026 14:27:10 +0800 Subject: [PATCH 4/4] Avoid PowerMock reflect dependency in LOAD tests --- .../load/LoadTsFileAnalyzerDiagnosticsTest.java | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) 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 f3f71af8710ea..6fe20c64bc2d1 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 @@ -53,10 +53,10 @@ 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.FileOutputStream; +import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.Collections; import java.util.Optional; @@ -258,8 +258,13 @@ private TSStatus analyzeAfterSchemaFailure(final RuntimeException exception) thr new MPPQueryContext(new QueryId("load_schema_recovery")), Mockito.mock(IPartitionFetcher.class), Mockito.mock(ISchemaFetcher.class))) { - final Object verifier = Whitebox.getInternalState(analyzer, "schemaAutoCreatorAndVerifier"); - final Object schemaCache = Whitebox.getInternalState(verifier, "schemaCache"); + final Field verifierField = + analyzer.getClass().getDeclaredField("schemaAutoCreatorAndVerifier"); + verifierField.setAccessible(true); + final Object verifier = verifierField.get(analyzer); + final Field schemaCacheField = verifier.getClass().getDeclaredField("schemaCache"); + schemaCacheField.setAccessible(true); + final Object schemaCache = schemaCacheField.get(verifier); final IDeviceID device = new PlainDeviceID("root.sg.d"); final Method addTimeSeries = schemaCache