diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBQueryWithCorruptedTsFileIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBQueryWithCorruptedTsFileIT.java new file mode 100644 index 0000000000000..ae2fcb039930c --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBQueryWithCorruptedTsFileIT.java @@ -0,0 +1,325 @@ +/* + * 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.relational.it.query.recent; + +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.exception.write.WriteProcessException; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.record.Tablet; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.apache.iotdb.db.it.utils.TestUtils.tableAssertTestFail; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +@RunWith(IoTDBTestRunner.class) +@Category({TableLocalStandaloneIT.class}) +public class IoTDBQueryWithCorruptedTsFileIT { + private static final String DATABASE_NAME = "test_corrupted_read_tsfile"; + + private static File tmpDir; + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv().initClusterEnvironment(); + try (Connection connection = EnvFactory.getEnv().getTableConnection(); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE " + DATABASE_NAME); + } + } + + @Before + public void setUpBeforeTest() throws IOException { + tmpDir = new File(Files.createTempDirectory("corrupt-tsfile").toUri()); + } + + @After + public void tearDownAfterTest() { + deleteTmpDir(); + } + + @AfterClass + public static void tearDown() { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testReadTsFileWithCorruptedMetadataIndexNode() throws Exception { + File tsFile = new File(tmpDir, "corrupt-meta.tsfile"); + try (TsFileWriter writer = new TsFileWriter(tsFile)) { + generateTable( + writer, "table1", Arrays.asList("tag1"), Arrays.asList("s1"), TSDataType.INT64, 1, 10); + } + + // TsFile layout: [Header] [Data] [MetadataIndex Tree] [TsFileMetadata] [Magic][Size] + // ↑ metaOffset ↑ fileMetadataPos + // + // MetadataIndexNode serialization: + // [entryCount (varInt)] [entry1]...[entryN] [endOffset (long, 8B)] [nodeType (1B)] + // nodeType valid values: 0=INTERNAL_DEVICE, 1=LEAF_DEVICE, 2=INTERNAL_MEASUREMENT, + // 3=LEAF_MEASUREMENT + // nodeType is the LAST byte before TsFileMetadata, i.e. at fileMetadataPos - 1 + long metaOffset; + long fileMetadataPos; + try (TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) { + metaOffset = reader.readFileMetadata().getMetaOffset(); + fileMetadataPos = reader.getFileMetadataPos(); + } + + // Corrupt the nodeType byte to 0xFF (all valid types are 0-3) + byte[] fileBytes = Files.readAllBytes(tsFile.toPath()); + fileBytes[(int) fileMetadataPos - 1] = (byte) 0xFF; + Files.write(tsFile.toPath(), fileBytes); + + tableAssertTestFail( + "SELECT * FROM read_tsfile(PATHS => '" + toSqlPath(tsFile) + "')", + "timeseries metadata", + DATABASE_NAME); + } + + @Test + public void testReadTsFileWithCorruptedPageData() throws Exception { + File tsFile = new File(tmpDir, "corrupt-page.tsfile"); + try (TsFileWriter writer = new TsFileWriter(tsFile)) { + generateTable( + writer, "table1", Arrays.asList("tag1"), Arrays.asList("s1"), TSDataType.INT64, 1, 100); + } + + corruptDataSection(tsFile); + + tableAssertTestFail( + "SELECT * FROM read_tsfile(PATHS => '" + toSqlPath(tsFile) + "')", "TsFile", DATABASE_NAME); + } + + @Test + public void testNormalQueryWithCorruptedPageData() throws Exception { + String tableName = "corrupt_table"; + // 1. Create table and insert data via session — generates TsFiles in the data directory + try (Connection connection = EnvFactory.getEnv().getTableConnection(); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE_NAME); + statement.execute( + "CREATE TABLE " + tableName + "(device_id STRING TAG, s1 INT64 FIELD, s2 INT64 FIELD)"); + for (int i = 1; i <= 200; i++) { + statement.execute( + "INSERT INTO " + + tableName + + "(time, device_id, s1, s2) VALUES(" + + i + + ", 'd" + + (i % 10) + + "', " + + i + + ", " + + (i * 10) + + ")"); + } + statement.execute("FLUSH"); + } + + // 2. Find the generated TsFile in the data directory + File sequenceDir = + new File( + EnvFactory.getEnv().getDataNodeWrapper(0).getDataNodeDir() + + File.separator + + "data" + + File.separator + + "sequence"); + File tsFile = findTsFileRecursively(sequenceDir); + if (tsFile == null) { + fail("Could not find TsFile in data directory: " + sequenceDir.getAbsolutePath()); + } + + // 3. Corrupt the data section of the TsFile + corruptDataSection(tsFile); + + // 4. Query — should fail with a corruption message that does NOT include the file path + try (Connection connection = EnvFactory.getEnv().getTableConnection(); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE_NAME); + try { + statement.execute("SELECT * FROM " + tableName + " ORDER BY time"); + fail("Expected query on corrupted TsFile to fail"); + } catch (SQLException e) { + assertTrue( + "Error message should mention corruption without file path: " + e.getMessage(), + e.getMessage().contains("may be corrupted") + || e.getMessage().contains("check the logs")); + } + } + } + + /** + * Corrupts a block of bytes in the data section of a TsFile (between the header and the + * MetadataIndex tree). Uses {@link TsFileSequenceReader} to read {@code metaOffset} from + * TsFileMetadata, so corruption reliably hits compressed page data rather than metadata. + * + *

TsFile layout: [Header] [Data chunks] [MetadataIndex Tree] [TsFileMetadata] [Magic][Size] ↑ + * metaOffset + */ + private static void corruptDataSection(File tsFile) throws IOException { + long metaOffset; + try (TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) { + metaOffset = reader.readFileMetadata().getMetaOffset(); + } + + byte[] fileBytes = Files.readAllBytes(tsFile.toPath()); + int magicLen = TSFileConfig.MAGIC_STRING.getBytes().length; + int dataStart = magicLen + Byte.BYTES; + int dataEnd = (int) metaOffset; + // Corrupt bytes in the middle of the data section — XOR 512 bytes to ensure decompression fails + int middle = dataStart + (dataEnd - dataStart) / 2; + int corruptLen = Math.min(512, dataEnd - middle); + for (int i = 0; i < corruptLen; i++) { + fileBytes[middle + i] ^= 0xFF; + } + Files.write(tsFile.toPath(), fileBytes); + } + + private static File findTsFileRecursively(File dir) { + if (dir == null || !dir.exists()) { + return null; + } + File[] files = dir.listFiles(); + if (files == null) { + return null; + } + for (File file : files) { + if (file.isDirectory()) { + File found = findTsFileRecursively(file); + if (found != null) { + return found; + } + } else if (file.getName().endsWith(".tsfile") && file.length() > 0) { + return file; + } + } + return null; + } + + private static void generateTable( + TsFileWriter writer, + String tableName, + List tagColumns, + List fieldColumns, + TSDataType fieldType, + long startTime, + long endTime) + throws IOException, WriteProcessException { + List columnNames = new ArrayList<>(tagColumns.size() + fieldColumns.size()); + List columnTypes = new ArrayList<>(tagColumns.size() + fieldColumns.size()); + List columnCategories = + new ArrayList<>(tagColumns.size() + fieldColumns.size()); + for (String tagColumn : tagColumns) { + columnNames.add(tagColumn); + columnTypes.add(TSDataType.STRING); + columnCategories.add(ColumnCategory.TAG); + } + for (String fieldColumn : fieldColumns) { + columnNames.add(fieldColumn); + columnTypes.add(fieldType); + columnCategories.add(ColumnCategory.FIELD); + } + + writer.registerTableSchema( + new TableSchema(tableName, columnNames, columnTypes, columnCategories)); + Tablet tablet = new Tablet(tableName, columnNames, columnTypes, columnCategories); + for (int deviceIndex = 1; deviceIndex <= 2; deviceIndex++) { + for (long time = startTime; time <= endTime; time++) { + int row = tablet.getRowSize(); + tablet.addTimestamp(row, time); + for (int i = 0; i < tagColumns.size(); i++) { + tablet.addValue(row, i, tagColumns.get(i) + "_" + deviceIndex); + } + for (int i = 0; i < fieldColumns.size(); i++) { + tablet.addValue(row, tagColumns.size() + i, time); + } + if (tablet.getRowSize() == tablet.getMaxRowNumber()) { + writer.writeTable(tablet); + tablet.reset(); + } + } + } + if (tablet.getRowSize() != 0) { + writer.writeTable(tablet); + } + } + + private static String toSqlPath(File file) { + return file.getAbsolutePath().replace("\\", "\\\\").replace("'", "''"); + } + + private static void deleteTmpDir() { + if (tmpDir == null || !tmpDir.exists()) { + return; + } + File[] files = tmpDir.listFiles(); + if (files != null) { + for (File file : files) { + deleteRecursively(file); + } + } + try { + Files.delete(tmpDir.toPath()); + } catch (IOException ignored) { + // ignore + } + } + + private static void deleteRecursively(File file) { + if (file.isDirectory()) { + File[] children = file.listFiles(); + if (children != null) { + for (File child : children) { + deleteRecursively(child); + } + } + } + try { + Files.delete(file.toPath()); + } catch (IOException ignored) { + // ignore + } + } +} 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 92242b62aa726..5839542d1b3d4 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 @@ -218,6 +218,32 @@ public final class DataNodeQueryMessages { "Error happened while scanning the file"; public static final String ALL_CACHED_CHUNKS_SHOULD_BE_CONSUMED_FIRST = "all cached chunks should be consumed first"; + public static final String + EXCEPTION_FAILED_TO_READ_TIMESERIES_METADATA_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_0B9E652E = + "Failed to read timeseries metadata. The TsFile may be corrupted," + + " please check the logs for the corrupted file path."; + public static final String EXCEPTION_FAILED_TO_READ_TIMESERIES_METADATA_FROM_TSFILE_ARG_B07568F8 = + "Failed to read timeseries metadata from TsFile: %s"; + public static final String + EXCEPTION_FAILED_TO_READ_CHUNK_DATA_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_F0FFE629 = + "Failed to read chunk data. The TsFile may be corrupted," + + " please check the logs for the corrupted file path."; + public static final String EXCEPTION_FAILED_TO_READ_CHUNK_DATA_FROM_TSFILE_ARG_B88F2496 = + "Failed to read chunk data from TsFile: %s"; + public static final String + EXCEPTION_FAILED_TO_DECODE_PAGE_DATA_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_54D7C6D9 = + "Failed to decode page data. The TsFile may be corrupted," + + " please check the logs for the corrupted file path."; + public static final String EXCEPTION_FAILED_TO_DECODE_PAGE_DATA_FROM_TSFILE_ARG_645F5377 = + "Failed to decode page data from TsFile: %s"; + public static final String + EXCEPTION_FAILED_TO_LOAD_PAGE_READER_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_05D35760 = + "Failed to load page reader. The TsFile may be corrupted," + + " please check the logs for the corrupted file path."; + public static final String EXCEPTION_FAILED_TO_LOAD_PAGE_READER_FROM_TSFILE_ARG_3B1CCC18 = + "Failed to load page reader from TsFile: %s"; + public static final String EXCEPTION_FAILED_TO_READ_METADATA_INDEX_NODE_FROM_TSFILE_ARG_EC5B6633 = + "Failed to read metadata index node from TsFile: %s"; public static final String OVERLAPPED_DATA_SHOULD_BE_CONSUMED_FIRST = "overlapped data should be consumed first"; public static final String NO_MORE_BATCH_DATA = 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 9ce80200a7b75..2ba8eb4162b84 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 @@ -205,6 +205,28 @@ public final class DataNodeQueryMessages { "扫描文件时发生错误"; public static final String ERROR_HAPPENED_WHILE_SCANNING_THE_FILE = "扫描文件时发生错误"; + public static final String + EXCEPTION_FAILED_TO_READ_TIMESERIES_METADATA_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_0B9E652E = + "读取时间序列元数据失败。TsFile 可能已损坏，请检查日志中的损坏文件路径。"; + public static final String EXCEPTION_FAILED_TO_READ_TIMESERIES_METADATA_FROM_TSFILE_ARG_B07568F8 = + "从 TsFile 读取时间序列元数据失败：%s"; + public static final String + EXCEPTION_FAILED_TO_READ_CHUNK_DATA_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_F0FFE629 = + "读取 chunk 数据失败。TsFile 可能已损坏，请检查日志中的损坏文件路径。"; + public static final String EXCEPTION_FAILED_TO_READ_CHUNK_DATA_FROM_TSFILE_ARG_B88F2496 = + "从 TsFile 读取 chunk 数据失败：%s"; + public static final String + EXCEPTION_FAILED_TO_DECODE_PAGE_DATA_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_54D7C6D9 = + "解码 page 数据失败。TsFile 可能已损坏，请检查日志中的损坏文件路径。"; + public static final String EXCEPTION_FAILED_TO_DECODE_PAGE_DATA_FROM_TSFILE_ARG_645F5377 = + "从 TsFile 解码 page 数据失败：%s"; + public static final String + EXCEPTION_FAILED_TO_LOAD_PAGE_READER_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_05D35760 = + "加载 page reader 失败。TsFile 可能已损坏，请检查日志中的损坏文件路径。"; + public static final String EXCEPTION_FAILED_TO_LOAD_PAGE_READER_FROM_TSFILE_ARG_3B1CCC18 = + "从 TsFile 加载 page reader 失败：%s"; + public static final String EXCEPTION_FAILED_TO_READ_METADATA_INDEX_NODE_FROM_TSFILE_ARG_EC5B6633 = + "从 TsFile 读取元数据索引节点失败：%s"; public static final String ALL_CACHED_CHUNKS_SHOULD_BE_CONSUMED_FIRST = "所有缓存的 chunk 应先被消费"; public static final String OVERLAPPED_DATA_SHOULD_BE_CONSUMED_FIRST = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/CorruptedTsFileException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/CorruptedTsFileException.java new file mode 100644 index 0000000000000..6574b3a042dc8 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/CorruptedTsFileException.java @@ -0,0 +1,74 @@ +/* + * 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; + +import org.apache.tsfile.exception.TsFileRuntimeException; + +import java.io.File; + +/** + * Thrown when a TsFile is detected to be corrupted during query execution. Extends {@link + * TsFileRuntimeException} so it follows the existing TsFile error handling and bypasses all {@code + * catch (IOException)} blocks in the operator hierarchy. + */ +public class CorruptedTsFileException extends TsFileRuntimeException { + + public enum Stage { + READ_TIMESERIES_METADATA, + READ_CHUNK_DATA, + LOAD_PAGE_READER, + DECODE_PAGE_DATA, + READ_METADATA_INDEX_NODE + } + + private final File tsFile; + private final Stage stage; + + /** + * Creates a CorruptedTsFileException. + * + *

The original exception {@code cause} is added via {@link #addSuppressed(Throwable)} rather + * than {@link #initCause(Throwable)}. This ensures that {@code + * ErrorHandlingCommonUtils.getRootCause(this)} returns this exception itself (not the wrapped + * IOException), so upstream error handling in {@code AbstractDriverThread} matches it correctly. + * The original cause stack trace is preserved via suppressed exceptions. + * + * @param tsFile the corrupted TsFile + * @param stage the operation that encountered the corruption + * @param message user-facing error message + * @param cause the original exception (preserved as suppressed) + */ + public CorruptedTsFileException(File tsFile, Stage stage, String message, Throwable cause) { + super(message); + this.tsFile = tsFile; + this.stage = stage; + if (cause != null) { + addSuppressed(cause); + } + } + + public File getTsFile() { + return tsFile; + } + + public Stage getStage() { + return stage; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java index 2a0aa01f6d9a4..810d9f54c62df 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.path.AlignedFullPath; import org.apache.iotdb.commons.path.NonAlignedFullPath; import org.apache.iotdb.db.exception.ChunkTypeInconsistentException; +import org.apache.iotdb.db.exception.CorruptedTsFileException; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; @@ -54,6 +55,7 @@ import org.apache.tsfile.read.reader.IPageReader; import org.apache.tsfile.write.schema.IMeasurementSchema; +import java.io.File; import java.io.IOException; import java.util.ArrayList; import java.util.Collections; @@ -163,6 +165,18 @@ public static TimeseriesMetadata loadTimeSeriesMetadata( } return timeSeriesMetadata; + } catch (Exception e) { + throw new CorruptedTsFileException( + resource.getTsFile(), + CorruptedTsFileException.Stage.READ_TIMESERIES_METADATA, + context.isExternalTsFileScan() + ? String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_READ_TIMESERIES_METADATA_FROM_TSFILE_ARG_B07568F8, + resource.getTsFile()) + : DataNodeQueryMessages + .EXCEPTION_FAILED_TO_READ_TIMESERIES_METADATA_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_0B9E652E, + e); } finally { long costTime = System.nanoTime() - t1; if (loadFromMem) { @@ -250,6 +264,18 @@ public static AbstractAlignedTimeSeriesMetadata loadAlignedTimeSeriesMetadata( } } return alignedTimeSeriesMetadata; + } catch (Exception e) { + throw new CorruptedTsFileException( + resource.getTsFile(), + CorruptedTsFileException.Stage.READ_TIMESERIES_METADATA, + context.isExternalTsFileScan() + ? String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_READ_TIMESERIES_METADATA_FROM_TSFILE_ARG_B07568F8, + resource.getTsFile()) + : DataNodeQueryMessages + .EXCEPTION_FAILED_TO_READ_TIMESERIES_METADATA_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_0B9E652E, + e); } finally { long costTime = System.nanoTime() - t1; if (loadFromMem) { @@ -469,25 +495,66 @@ public static List loadChunkMetadataList(ITimeSeriesMetadata tim * IOException will be thrown */ public static List loadPageReaderList( - IChunkMetadata chunkMetaData, Filter globalTimeFilter, List targetDataTypeList) + IChunkMetadata chunkMetaData, + Filter globalTimeFilter, + List targetDataTypeList, + FragmentInstanceContext context) throws IOException { checkArgument( chunkMetaData != null, DataNodeQueryMessages.EXCEPTION_CAN_QUOTE_T_INIT_NULL_CHUNKMETA_15C12BEE); IChunkLoader chunkLoader = chunkMetaData.getChunkLoader(); - IChunkReader chunkReader; + File tsFile = null; + if (chunkLoader instanceof DiskChunkLoader) { + tsFile = ((DiskChunkLoader) chunkLoader).getTsFile(); + } else if (chunkLoader instanceof DiskAlignedChunkLoader) { + tsFile = ((DiskAlignedChunkLoader) chunkLoader).getTsFile(); + } + final IChunkReader chunkReader; try { chunkReader = chunkLoader.getChunkReader(chunkMetaData, globalTimeFilter); } catch (ChunkTypeInconsistentException e) { // if the chunk in tsfile is a value chunk of aligned series but registered series is // non-aligned, we should skip all data of this chunk. return Collections.emptyList(); + } catch (Exception e) { + if (tsFile == null) { + throw e; + } + throw new CorruptedTsFileException( + tsFile, + CorruptedTsFileException.Stage.READ_CHUNK_DATA, + context.isExternalTsFileScan() + ? String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_READ_CHUNK_DATA_FROM_TSFILE_ARG_B88F2496, + tsFile) + : DataNodeQueryMessages + .EXCEPTION_FAILED_TO_READ_CHUNK_DATA_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_F0FFE629, + e); } if (chunkMetaData.isDataTypeModifiedAndCannotUseStatistics()) { chunkReader.markDataTypeModifiedAndCannotUseStatistics(); } - return chunkReader.loadPageReaderList(); + try { + return chunkReader.loadPageReaderList(); + } catch (Exception e) { + if (tsFile == null) { + throw e; + } + throw new CorruptedTsFileException( + tsFile, + CorruptedTsFileException.Stage.LOAD_PAGE_READER, + context.isExternalTsFileScan() + ? String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_LOAD_PAGE_READER_FROM_TSFILE_ARG_3B1CCC18, + tsFile) + : DataNodeQueryMessages + .EXCEPTION_FAILED_TO_LOAD_PAGE_READER_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_05D35760, + e); + } } /** diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java index 52ddf505a5e54..1b8ca2b1020c1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java @@ -22,6 +22,7 @@ import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; import org.apache.iotdb.commons.path.IFullPath; import org.apache.iotdb.commons.path.NonAlignedFullPath; +import org.apache.iotdb.db.exception.CorruptedTsFileException; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; @@ -32,6 +33,8 @@ import org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedReadOnlyMemChunk; import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.DiskAlignedChunkLoader; +import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.DiskChunkLoader; import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemAlignedPageReader; import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemChunkLoader; import org.apache.iotdb.db.storageengine.dataregion.read.reader.chunk.MemPageReader; @@ -79,6 +82,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.File; import java.io.IOException; import java.io.Serializable; import java.nio.charset.StandardCharsets; @@ -695,6 +699,14 @@ private void unpackOneChunkMetaData(IChunkMetadata chunkMetaData) throws IOExcep long timestampInFileName = FileLoaderUtils.getTimestampInFileName(chunkMetaData); IChunkLoader chunkLoader = chunkMetaData.getChunkLoader(); + final File tsFile; + if (chunkLoader instanceof DiskChunkLoader) { + tsFile = ((DiskChunkLoader) chunkLoader).getTsFile(); + } else if (chunkLoader instanceof DiskAlignedChunkLoader) { + tsFile = ((DiskAlignedChunkLoader) chunkLoader).getTsFile(); + } else { + tsFile = null; + } if ((chunkLoader instanceof MemChunkLoader) && ((MemChunkLoader) chunkLoader).isStreamingQueryMemChunk()) { unpackOneFakeMemChunkMetaData( @@ -703,7 +715,7 @@ private void unpackOneChunkMetaData(IChunkMetadata chunkMetaData) throws IOExcep } List pageReaderList = FileLoaderUtils.loadPageReaderList( - chunkMetaData, scanOptions.getGlobalTimeFilter(), getTsDataTypeList()); + chunkMetaData, scanOptions.getGlobalTimeFilter(), getTsDataTypeList(), context); // init TsBlockBuilder for each page reader pageReaderList.forEach(p -> p.initTsBlockBuilder(getTsDataTypeList())); @@ -718,7 +730,8 @@ private void unpackOneChunkMetaData(IChunkMetadata chunkMetaData) throws IOExcep chunkMetaData.getVersion(), chunkMetaData.getOffsetOfChunkHeader(), iPageReader, - true)); + true, + tsFile)); } } else { for (int i = pageReaderList.size() - 1; i >= 0; i--) { @@ -729,7 +742,8 @@ private void unpackOneChunkMetaData(IChunkMetadata chunkMetaData) throws IOExcep chunkMetaData.getVersion(), chunkMetaData.getOffsetOfChunkHeader(), pageReaderList.get(i), - true)); + true, + tsFile)); } } } else { @@ -742,7 +756,8 @@ private void unpackOneChunkMetaData(IChunkMetadata chunkMetaData) throws IOExcep chunkMetaData.getVersion(), chunkMetaData.getOffsetOfChunkHeader(), pageReader, - false))); + false, + tsFile))); } if (LOGGER.isDebugEnabled()) { @@ -2083,6 +2098,7 @@ protected static class VersionPageReader implements IVersionPageReader { protected final boolean isSeq; protected final boolean isAligned; protected final boolean isMem; + protected final File tsFile; VersionPageReader( QueryContext context, @@ -2090,7 +2106,8 @@ protected static class VersionPageReader implements IVersionPageReader { long version, long offset, IPageReader data, - boolean isSeq) { + boolean isSeq, + File tsFile) { this.context = context; this.version = new MergeReaderPriority(fileTimestamp, version, offset, isSeq); this.data = data; @@ -2100,6 +2117,7 @@ protected static class VersionPageReader implements IVersionPageReader { || data instanceof MemAlignedPageReader || data instanceof TablePageReader; this.isMem = data instanceof MemPageReader || data instanceof MemAlignedPageReader; + this.tsFile = tsFile; } @SuppressWarnings("squid:S3740") @@ -2144,6 +2162,21 @@ public TsBlock getAllSatisfiedPageData(boolean ascending) throws IOException { CommonUtils.toString(tsBlock)); } return tsBlock; + } catch (Exception e) { + if (tsFile != null) { + throw new CorruptedTsFileException( + tsFile, + CorruptedTsFileException.Stage.DECODE_PAGE_DATA, + context.isExternalTsFileScan() + ? String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_DECODE_PAGE_DATA_FROM_TSFILE_ARG_645F5377, + tsFile) + : DataNodeQueryMessages + .EXCEPTION_FAILED_TO_DECODE_PAGE_DATA_THE_TSFILE_MAY_BE_CORRUPTED_PLEASE_CHECK_THE_LOGS_FOR_THE_CORRUPTED_FILE_PATH_54D7C6D9, + e); + } + throw e; } finally { long time = System.nanoTime() - startTime; if (isAligned) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThread.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThread.java index 4fb9e8a8cb5ca..7735ecae3e8f7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThread.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/AbstractDriverThread.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.exception.IoTDBException; import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils; +import org.apache.iotdb.db.exception.CorruptedTsFileException; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.execution.schedule.task.DriverTask; import org.apache.iotdb.db.utils.SetThreadName; @@ -93,6 +94,12 @@ public void run() { next.setAbortCause( new IoTDBRuntimeException( rootCause.getMessage(), DATE_OUT_OF_RANGE.getStatusCode(), true)); + } else if (rootCause instanceof CorruptedTsFileException) { + // CorruptedTsFileException no longer chains the original IOException as its + // cause (it uses addSuppressed instead), so getRootCause returns the exception + // itself and we can match it here. + logger.warn(DataNodeQueryMessages.EXECUTEFAILED, rootCause); + next.setAbortCause(rootCause); } else { logger.warn(DataNodeQueryMessages.EXECUTEFAILED, rootCause); next.setAbortCause( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java index d16e602e3d23f..9a96145bc8a0b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java @@ -28,6 +28,7 @@ import org.apache.iotdb.commons.schema.filter.SchemaFilter; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.exception.CorruptedTsFileException; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.common.QueryId; @@ -671,24 +672,45 @@ private class DeviceCollector implements Closeable { private DeviceCollector() { try { for (int fileIndex = 0; fileIndex < tsFilePaths.size(); fileIndex++) { - TsFileSequenceReader reader = - FileReaderManager.getInstance() - .get(tsFilePaths.get(fileIndex), null, true, null, true); - deviceIteratorMap.put(fileIndex, new LazyTsFileDeviceIterator(reader, tableName, null)); + try { + TsFileSequenceReader reader = + FileReaderManager.getInstance() + .get(tsFilePaths.get(fileIndex), null, true, null, true); + deviceIteratorMap.put(fileIndex, new LazyTsFileDeviceIterator(reader, tableName, null)); + } catch (Exception e) { + throw corruptedMetadataIndexNodeException(fileIndex, e); + } } - } catch (IOException e) { + } catch (RuntimeException e) { close(); - throw new RuntimeException( - DataNodeQueryMessages.FAILED_TO_CREATE_EXTERNAL_TSFILE_DEVICE_COLLECTOR, e); + throw e; } } + private CorruptedTsFileException corruptedMetadataIndexNodeException( + int fileIndex, Exception cause) { + File tsFile = sharedTsFileResources.get(fileIndex).getTsFile(); + return new CorruptedTsFileException( + tsFile, + CorruptedTsFileException.Stage.READ_METADATA_INDEX_NODE, + String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_READ_METADATA_INDEX_NODE_FROM_TSFILE_ARG_EC5B6633, + tsFile), + cause); + } + private boolean hasNextDevice() { - for (LazyTsFileDeviceIterator deviceIterator : deviceIteratorMap.values()) { - if (deviceIterator.hasNext() - || (deviceIterator.hasCurrent() - && !deviceIterator.getCurrentDeviceID().equals(currentDevice))) { - return true; + for (Map.Entry entry : deviceIteratorMap.entrySet()) { + try { + LazyTsFileDeviceIterator deviceIterator = entry.getValue(); + if (deviceIterator.hasNext() + || (deviceIterator.hasCurrent() + && !deviceIterator.getCurrentDeviceID().equals(currentDevice))) { + return true; + } + } catch (Exception e) { + throw corruptedMetadataIndexNodeException(entry.getKey(), e); } } return false; @@ -700,21 +722,25 @@ private IDeviceID nextDevice() { deviceIteratorMap.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry entry = iterator.next(); - LazyTsFileDeviceIterator deviceIterator = entry.getValue(); - IDeviceID currentFileDevice = null; - if (deviceIterator.hasCurrent()) { - currentFileDevice = deviceIterator.getCurrentDeviceID(); - } - if (currentFileDevice == null || currentFileDevice.equals(currentDevice)) { - if (deviceIterator.hasNext()) { - currentFileDevice = deviceIterator.next(); - } else { - iterator.remove(); - continue; + try { + LazyTsFileDeviceIterator deviceIterator = entry.getValue(); + IDeviceID currentFileDevice = null; + if (deviceIterator.hasCurrent()) { + currentFileDevice = deviceIterator.getCurrentDeviceID(); } - } - if (minDevice == null || minDevice.compareTo(currentFileDevice) > 0) { - minDevice = currentFileDevice; + if (currentFileDevice == null || currentFileDevice.equals(currentDevice)) { + if (deviceIterator.hasNext()) { + currentFileDevice = deviceIterator.next(); + } else { + iterator.remove(); + continue; + } + } + if (minDevice == null || minDevice.compareTo(currentFileDevice) > 0) { + minDevice = currentFileDevice; + } + } catch (Exception e) { + throw corruptedMetadataIndexNodeException(entry.getKey(), e); } } currentDevice = minDevice; @@ -725,15 +751,19 @@ private IDeviceID nextDevice() { private void collectCurrentDeviceOffsets() { List deviceOffsets = new ArrayList<>(); for (Map.Entry entry : deviceIteratorMap.entrySet()) { - LazyTsFileDeviceIterator deviceIterator = entry.getValue(); - if (currentDevice != null - && deviceIterator.hasCurrent() - && currentDevice.equals(deviceIterator.getCurrentDeviceID())) { - deviceOffsets.add( - new ExternalTsFileDeviceQueryTask.DeviceOffset( - entry.getKey(), - deviceIterator.getCurrentDeviceMeasurementNodeOffset()[0], - deviceIterator.getCurrentDeviceMeasurementNodeOffset()[1])); + try { + LazyTsFileDeviceIterator deviceIterator = entry.getValue(); + if (currentDevice != null + && deviceIterator.hasCurrent() + && currentDevice.equals(deviceIterator.getCurrentDeviceID())) { + deviceOffsets.add( + new ExternalTsFileDeviceQueryTask.DeviceOffset( + entry.getKey(), + deviceIterator.getCurrentDeviceMeasurementNodeOffset()[0], + deviceIterator.getCurrentDeviceMeasurementNodeOffset()[1])); + } + } catch (Exception e) { + throw corruptedMetadataIndexNodeException(entry.getKey(), e); } } currentDeviceOffsets = deviceOffsets; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java index 7c1e9d1263518..5af552502857a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java @@ -37,6 +37,7 @@ import org.apache.tsfile.read.reader.chunk.AlignedChunkReader; import org.apache.tsfile.read.reader.chunk.TableChunkReader; +import java.io.File; import java.io.IOException; import java.util.ArrayList; import java.util.List; @@ -142,4 +143,8 @@ public IChunkReader getChunkReader(IChunkMetadata chunkMetaData, Filter globalTi public TsFileID getTsFileID() { return resource.getTsFileID(); } + + public File getTsFile() { + return resource.getTsFile(); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskChunkLoader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskChunkLoader.java index 7d259a9e50534..eb7abb7ed07f6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskChunkLoader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskChunkLoader.java @@ -37,6 +37,7 @@ import org.apache.tsfile.read.reader.IChunkReader; import org.apache.tsfile.read.reader.chunk.ChunkReader; +import java.io.File; import java.io.IOException; import static org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet.INIT_CHUNK_READER_NONALIGNED_DISK; @@ -122,4 +123,8 @@ public IChunkReader getChunkReader(IChunkMetadata chunkMetaData, Filter globalTi public TsFileID getTsFileID() { return resource.getTsFileID(); } + + public File getTsFile() { + return resource.getTsFile(); + } }