From f97e4460687dab61f767e425ce61084b1f505a48 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 10 Sep 2026 17:34:59 +0800 Subject: [PATCH] [Load] Preserve invalid device path failures during tablet conversion (#18614) (cherry picked from commit 6a3c8d667f0552966f4ccf7e660c5ea356e8985d) --- .../LoadAnalyzeInvalidPathException.java | 28 +++ .../plan/analyze/load/LoadTsFileAnalyzer.java | 28 +-- .../load/LoadTsFilePathUtils.java | 47 ++++ .../LoadTreeTsFileTabletIterator.java | 26 ++- .../analyze/load/LoadTsFileAnalyzerTest.java | 11 +- .../converter/LoadTsFileInvalidPathTest.java | 205 ++++++++++++++++++ 6 files changed, 328 insertions(+), 17 deletions(-) create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadAnalyzeInvalidPathException.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePathUtils.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileInvalidPathTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadAnalyzeInvalidPathException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadAnalyzeInvalidPathException.java new file mode 100644 index 000000000000..ed8182008702 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/LoadAnalyzeInvalidPathException.java @@ -0,0 +1,28 @@ +/* + * 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.exception; + +/** An invalid tree path cannot be repaired by converting the TsFile to tablets. */ +public class LoadAnalyzeInvalidPathException extends LoadAnalyzeException { + + public LoadAnalyzeInvalidPathException(String message) { + super(message); + } +} 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 1715fe4bb271..3e25e033626e 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 @@ -37,6 +37,7 @@ import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.LoadAnalyzeException; +import org.apache.iotdb.db.exception.LoadAnalyzeInvalidPathException; import org.apache.iotdb.db.exception.LoadAnalyzeMissingSchemaException; import org.apache.iotdb.db.exception.LoadAnalyzeTypeMismatchException; import org.apache.iotdb.db.exception.load.LoadEmptyFileException; @@ -111,6 +112,7 @@ import java.util.Set; import java.util.stream.Collectors; +import static org.apache.iotdb.db.storageengine.load.LoadTsFilePathUtils.getValidatedDevicePath; import static org.apache.iotdb.db.storageengine.load.metrics.LoadTsFileCostMetricsSet.ANALYSIS; import static org.apache.iotdb.db.storageengine.load.metrics.LoadTsFileCostMetricsSet.ANALYSIS_ASYNC_MOVE; @@ -611,8 +613,9 @@ boolean isTemporaryUnavailableDueToPipeSchemaNotReady(final Throwable throwable) } private boolean shouldSkipConversion(LoadAnalyzeException e) { - return (e instanceof LoadAnalyzeTypeMismatchException) - && !loadTsFileStatement.isConvertOnTypeMismatch(); + return e instanceof LoadAnalyzeInvalidPathException + || (e instanceof LoadAnalyzeTypeMismatchException) + && !loadTsFileStatement.isConvertOnTypeMismatch(); } @Override @@ -643,6 +646,8 @@ public void autoCreateAndVerify( device2TimeSeriesMetadataList.entrySet()) { final IDeviceID device = entry.getKey(); + getValidatedDevicePath(device); + try { if (schemaCache.isDeviceDeletedByMods(device)) { continue; @@ -706,11 +711,13 @@ public void autoCreateAndVerify( public void checkWritePermission( Map> device2TimeseriesMetadataList) - throws AuthException { + throws AuthException, LoadAnalyzeInvalidPathException { for (final Map.Entry> entry : device2TimeseriesMetadataList.entrySet()) { final IDeviceID device = entry.getKey(); + getValidatedDevicePath(device); + try { if (schemaCache.isDeviceDeletedByMods(device)) { continue; @@ -813,7 +820,9 @@ private void doAutoCreateAndVerify() if (isVerifySchema) { verifySchema(schemaTree); } - } catch (AuthException | LoadAnalyzeTypeMismatchException e) { + } catch (AuthException + | LoadAnalyzeInvalidPathException + | LoadAnalyzeTypeMismatchException e) { throw e; } catch (LoadAnalyzeMissingSchemaException e) { if (isTemporaryUnavailableDueToPipeSchemaNotReady(e)) { @@ -858,18 +867,9 @@ private void autoCreateDatabase() final Set databasesNeededToBeSet = new HashSet<>(); for (final IDeviceID device : schemaCache.getDevice2TimeSeries().keySet()) { - final PartialPath devicePath; - try { - devicePath = new PartialPath(device); - } catch (final IllegalPathException e) { - throw new LoadAnalyzeException(e.getMessage()); - } + final PartialPath devicePath = getValidatedDevicePath(device); final String[] devicePrefixNodes = devicePath.getNodes(); - if (hasEmptyPathNode(devicePath)) { - throw new LoadAnalyzeException( - new IllegalPathException(devicePath.getFullPath()).getMessage()); - } if (devicePrefixNodes.length < databasePrefixNodesLength) { throw new LoadAnalyzeException( String.format( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePathUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePathUtils.java new file mode 100644 index 000000000000..8d83746ce673 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFilePathUtils.java @@ -0,0 +1,47 @@ +/* + * 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.storageengine.load; + +import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.db.exception.LoadAnalyzeInvalidPathException; + +import org.apache.tsfile.file.metadata.IDeviceID; + +public class LoadTsFilePathUtils { + + private LoadTsFilePathUtils() {} + + public static PartialPath getValidatedDevicePath(final IDeviceID device) + throws LoadAnalyzeInvalidPathException { + try { + final PartialPath devicePath = new PartialPath(device); + // Validate the original nodes before converting to a string, which loses null nodes. + for (final String node : devicePath.getNodes()) { + if (node == null || node.isEmpty()) { + throw new IllegalPathException(devicePath.getFullPath()); + } + } + return devicePath; + } catch (final IllegalPathException e) { + throw new LoadAnalyzeInvalidPathException(e.getMessage()); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java index 41c88ce4752b..8c7e260aa680 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeTsFileTabletIterator.java @@ -19,9 +19,11 @@ package org.apache.iotdb.db.storageengine.load.converter; +import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.PipePattern; +import org.apache.iotdb.db.exception.LoadAnalyzeInvalidPathException; import org.apache.iotdb.db.exception.load.LoadRuntimeOutOfMemoryException; import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; import org.apache.iotdb.db.pipe.event.common.tsfile.container.query.TsFileInsertionQueryDataContainer; @@ -29,6 +31,7 @@ import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileParserMemoryManager; import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; +import org.apache.tsfile.exception.PathParseException; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.PlainDeviceID; import org.apache.tsfile.file.metadata.TimeseriesMetadata; @@ -54,6 +57,8 @@ import java.util.Set; import java.util.stream.Collectors; +import static org.apache.iotdb.db.storageengine.load.LoadTsFilePathUtils.getValidatedDevicePath; + /** * Load uses scan parsing first for throughput. If scan parsing hits corruption, fall back to query * parsing for the remaining measurements and devices so later data can still be loaded. @@ -63,7 +68,18 @@ class LoadTreeTsFileTabletIterator private static final Logger LOGGER = LoggerFactory.getLogger(LoadTreeTsFileTabletIterator.class); - private static final PipePattern LOAD_TREE_PATTERN = new IoTDBPipePattern(null); + private static final PipePattern LOAD_TREE_PATTERN = + new IoTDBPipePattern(null) { + @Override + public boolean mayOverlapWithDevice(final String device) { + try { + getValidatedDevicePath(new PlainDeviceID(device)); + } catch (final LoadAnalyzeInvalidPathException e) { + throw new IllegalArgumentException(e.getMessage(), e); + } + return super.mayOverlapWithDevice(device); + } + }; private final File file; private final boolean isWithMod; @@ -333,6 +349,7 @@ private boolean activateNextQueryParser() { while (!pendingQueryTasks.isEmpty()) { activeQueryTask = pendingQueryTasks.removeFirst(); try { + getValidatedDevicePath(activeQueryTask.device); activeQueryParser = new TsFileInsertionQueryDataContainer( file, @@ -412,6 +429,10 @@ private boolean shouldRethrow(final Exception e) { Throwable current = e; while (Objects.nonNull(current)) { if (current instanceof InterruptedException + // Invalid paths cannot be recovered by query parsing or splitting measurements. + || current instanceof PathParseException + || current instanceof IllegalPathException + || current instanceof LoadAnalyzeInvalidPathException || current instanceof PipeRuntimeOutOfMemoryCriticalException || current instanceof LoadRuntimeOutOfMemoryException) { return true; @@ -422,6 +443,9 @@ private boolean shouldRethrow(final Exception e) { } private RuntimeException toRuntimeException(final Exception e) { + if (e instanceof LoadAnalyzeInvalidPathException) { + return new IllegalArgumentException(e.getMessage(), e); + } return e instanceof RuntimeException ? (RuntimeException) e : new IllegalStateException("Failed to iterate tablets while loading TsFile.", e); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java index 589542c80230..0f3e2b191345 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerTest.java @@ -21,7 +21,7 @@ import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.conf.IoTDBDescriptor; -import org.apache.iotdb.db.exception.LoadAnalyzeException; +import org.apache.iotdb.db.exception.LoadAnalyzeInvalidPathException; import org.apache.iotdb.db.exception.LoadAnalyzeMissingSchemaException; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.common.QueryId; @@ -65,7 +65,7 @@ public void testSchemaVerifierShouldRejectDeviceWithEmptyPathNode() throws Excep () -> getAutoCreateDatabaseMethod(verifier).invoke(verifier)); Assert.assertTrue( String.valueOf(exception.getCause()), - exception.getCause() instanceof LoadAnalyzeException); + exception.getCause() instanceof LoadAnalyzeInvalidPathException); } finally { Assert.assertTrue(tsFile.delete()); } @@ -97,6 +97,13 @@ public void testSchemaVerifierShouldIgnoreLegacyDatabaseWithEmptyPathNode() thro Collections.singleton(databaseWithSameStringPrefix), databasesNeededToBeSet); Assert.assertEquals( Collections.singleton(database), getAlreadySetDatabases(getSchemaCache(verifier))); + + addTimeSeries( + getSchemaCache(verifier), + new PlainDeviceID("root.sg.d1"), + new MeasurementSchema("s1", TSDataType.INT32)); + // A valid device still uses its existing database despite the legacy root. entry. + getAutoCreateDatabaseMethod(verifier).invoke(verifier); } finally { Assert.assertTrue(tsFile.delete()); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileInvalidPathTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileInvalidPathTest.java new file mode 100644 index 000000000000..2775a432ba66 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileInvalidPathTest.java @@ -0,0 +1,205 @@ +/* + * 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.storageengine.load.converter; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.common.SessionInfo; +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.load.LoadTsFileAnalyzer; +import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; +import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileMemoryManager; +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.PlainDeviceID; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; +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 java.io.File; +import java.time.ZoneId; +import java.util.Arrays; +import java.util.Optional; + +import static org.mockito.Mockito.mock; + +public class LoadTsFileInvalidPathTest { + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private int dataNodeId; + + @Before + public void setUp() { + dataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId(); + IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0); + } + + @After + public void tearDown() { + IoTDBDescriptor.getInstance().getConfig().setDataNodeId(dataNodeId); + } + + @Test + public void testAnalysisRejectsEmptyDeviceNodeWithAndWithoutResource() throws Exception { + final IDeviceID device = new PlainDeviceID("root."); + final File file = writeTsFile(device); + assertAnalysisRejectsInvalidPath(file, device, true, true, -1); + + final TsFileResource resource = new TsFileResource(file); + resource.updateStartTime(device, 1); + resource.updateEndTime(device, 1); + resource.serialize(); + Assert.assertTrue(resource.resourceFileExists()); + assertAnalysisRejectsInvalidPath(file, device, true, true, -1); + } + + @Test + public void testAnalysisRejectsEmptyInnerDeviceNode() throws Exception { + final IDeviceID device = new PlainDeviceID("root.sg..d1"); + assertAnalysisRejectsInvalidPath(writeTsFile(device), device, true, true, -1); + } + + @Test + public void testAnalysisRejectsInvalidDeviceWithoutAutoCreateDatabase() throws Exception { + final IDeviceID device = new PlainDeviceID("root."); + assertAnalysisRejectsInvalidPath(writeTsFile(device), device, false, true, -1); + } + + @Test + public void testAnalysisRejectsInvalidDeviceWithoutSchemaChecks() throws Exception { + final IDeviceID device = new PlainDeviceID("root."); + assertAnalysisRejectsInvalidPath(writeTsFile(device), device, false, false, -1); + } + + @Test + public void testMiniFileAnalysisRejectsInvalidDevice() throws Exception { + final IDeviceID device = new PlainDeviceID("root."); + assertAnalysisRejectsInvalidPath(writeTsFile(device), device, true, true, Long.MAX_VALUE); + } + + @Test + public void testConversionReportsInvalidPathAndReleasesMemory() throws Exception { + assertConversionRejectsInvalidPath(new PlainDeviceID("root.")); + } + + @Test + public void testConversionRejectsEmptyInnerDeviceNode() throws Exception { + assertConversionRejectsInvalidPath(new PlainDeviceID("root.sg..d1")); + } + + private void assertConversionRejectsInvalidPath(final IDeviceID device) throws Exception { + final File file = writeTsFile(device); + final long memoryBefore = LoadTsFileMemoryManager.getInstance().getUsedMemorySizeInBytes(); + final LoadTreeStatementDataTypeConvertExecutionVisitor visitor = + new LoadTreeStatementDataTypeConvertExecutionVisitor( + statement -> { + Assert.fail("An invalid device must not be inserted."); + return null; + }); + + final Optional status = + visitor.visitLoadFile(LoadTsFileStatement.createUnchecked(file.getAbsolutePath()), null); + + Assert.assertTrue(status.isPresent()); + Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), status.get().getCode()); + Assert.assertNotNull(status.get().getMessage()); + Assert.assertEquals( + new IllegalPathException(((PlainDeviceID) device).toStringID()).getMessage(), + status.get().getMessage()); + Assert.assertEquals( + memoryBefore, LoadTsFileMemoryManager.getInstance().getUsedMemorySizeInBytes()); + } + + private void assertAnalysisRejectsInvalidPath( + final File file, + final IDeviceID device, + final boolean autoCreateDatabase, + final boolean checkSchema, + final long conversionThreshold) + throws Exception { + final LoadTsFileStatement statement = + LoadTsFileStatement.createUnchecked(file.getAbsolutePath()); + statement.setAutoCreateDatabase(autoCreateDatabase); + statement.setConvertOnTypeMismatch(true); + statement.setAutoCreateSchema(checkSchema); + statement.setVerifySchema(checkSchema); + statement.setTabletConversionThresholdBytes(conversionThreshold); + final MPPQueryContext context = + new MPPQueryContext( + "", + new QueryId("load_invalid_path_test"), + new SessionInfo(0, "root", ZoneId.systemDefault(), ""), + null, + null); + + try (final LoadTsFileAnalyzer analyzer = + new LoadTsFileAnalyzer( + statement, context, mock(IPartitionFetcher.class), mock(ISchemaFetcher.class))) { + final Analysis analysis = new Analysis(); + analyzer.analyzeFileByFile(analysis); + Assert.assertTrue(analysis.isFinishQueryAfterAnalyze()); + Assert.assertNotNull(analysis.getFailStatus()); + Assert.assertEquals( + TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), analysis.getFailStatus().getCode()); + Assert.assertEquals( + String.format( + "Loading file %s failed. Detail: %s", + file.getAbsolutePath(), + new IllegalPathException(((PlainDeviceID) device).toStringID()).getMessage()), + analysis.getFailStatus().getMessage()); + Assert.assertTrue(file.exists()); + } + } + + private File writeTsFile(final IDeviceID device) throws Exception { + final File file = new File(temporaryFolder.getRoot(), "1-1-0-0.tsfile"); + // Use the low-level writer to preserve legacy empty device nodes in the file. + try (final TsFileIOWriter writer = new TsFileIOWriter(file)) { + writer.startChunkGroup(device); + final AlignedChunkWriterImpl chunkWriter = + new AlignedChunkWriterImpl( + Arrays.asList( + new MeasurementSchema("quality", TSDataType.INT32), + new MeasurementSchema("value", TSDataType.DOUBLE))); + chunkWriter.getTimeChunkWriter().write(1); + chunkWriter.getValueChunkWriterByIndex(0).write(1, 1, false); + chunkWriter.getValueChunkWriterByIndex(1).write(1, 1.0, false); + chunkWriter.writeToFileWriter(writer); + writer.endChunkGroup(); + writer.endFile(); + } + return file; + } +}