This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch tsfile-corruption-error-message in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3c257889644a460d225275588a0165bd43979168 Author: shuwenwei <[email protected]> AuthorDate: Fri Jul 24 18:37:16 2026 +0800 Add CorruptedTsFileException for TsFile corruption errors with file path during query execution - Introduce CorruptedTsFileException extending RuntimeException with Stage enum (READ_TIMESERIES_METADATA, READ_CHUNK_DATA, LOAD_PAGE_READER, DECODE_PAGE_DATA, READ_METADATA_INDEX_NODE) - Uses super(message) + addSuppressed(cause) to prevent getRootCause penetration - Carries File reference and Stage for precise error reporting - Catch points in FileLoaderUtils, SeriesScanUtil, and DeviceCollector wrap IOException/RuntimeException as CorruptedTsFileException - Normal queries: message mentions corruption, tells user to check logs (no file path) - read_tsfile queries: message includes full TsFile path - DiskChunkLoader/DiskAlignedChunkLoader expose getTsFile() for File reference - ErrorHandlingUtils handles CorruptedTsFileException with TSFILE_PROCESSOR_ERROR - AbstractDriverThread matches CorruptedTsFileException by instanceof - AbstractDriverThread catch CorruptedTsFileException as abort cause - i18n: add EXCEPTION_ constants for all five stages (en + zh) - IT: IoTDBQueryWithCorruptedTsFileIT tests metadata index and page data corruption --- .../recent/IoTDBQueryWithCorruptedTsFileIT.java | 325 +++++++++++++++++++++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 26 ++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 22 ++ .../db/exception/CorruptedTsFileException.java | 72 +++++ .../execution/operator/source/FileLoaderUtils.java | 67 ++++- .../execution/operator/source/SeriesScanUtil.java | 43 ++- .../execution/schedule/AbstractDriverThread.java | 7 + .../read_tsfile/ExternalTsFileQueryResource.java | 106 ++++--- .../read/reader/chunk/DiskAlignedChunkLoader.java | 5 + .../read/reader/chunk/DiskChunkLoader.java | 5 + .../apache/iotdb/db/utils/ErrorHandlingUtils.java | 3 + 11 files changed, 638 insertions(+), 43 deletions(-) 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 00000000000..ae2fcb03993 --- /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. + * + * <p>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<String> tagColumns, + List<String> fieldColumns, + TSDataType fieldType, + long startTime, + long endTime) + throws IOException, WriteProcessException { + List<String> columnNames = new ArrayList<>(tagColumns.size() + fieldColumns.size()); + List<TSDataType> columnTypes = new ArrayList<>(tagColumns.size() + fieldColumns.size()); + List<ColumnCategory> 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 92242b62aa7..5839542d1b3 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 9ce80200a7b..2ba8eb4162b 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 00000000000..7a85307d8ad --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/CorruptedTsFileException.java @@ -0,0 +1,72 @@ +/* + * 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 java.io.File; + +/** + * Thrown when a TsFile is detected to be corrupted during query execution. Extends RuntimeException + * so it bypasses all {@code catch (IOException)} blocks in the operator hierarchy and reaches the + * centralized error handling in ErrorHandlingUtils. + */ +public class CorruptedTsFileException extends RuntimeException { + + 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. + * + * <p>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 2a0aa01f6d9..285017b4225 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 @@ package org.apache.iotdb.db.queryengine.execution.operator.source; 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.IChunkReader; 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 class FileLoaderUtils { } 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 class FileLoaderUtils { } } 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,60 @@ public class FileLoaderUtils { * IOException will be thrown */ public static List<IPageReader> loadPageReaderList( - IChunkMetadata chunkMetaData, Filter globalTimeFilter, List<TSDataType> targetDataTypeList) + IChunkMetadata chunkMetaData, + Filter globalTimeFilter, + List<TSDataType> 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) { + 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) { + 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 52ddf505a5e..1b8ca2b1020 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 @@ package org.apache.iotdb.db.queryengine.execution.operator.source; 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.queryengine.plan.statement.component.Ordering; 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.apache.tsfile.write.UnSupportedDataTypeException; 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 @@ public class SeriesScanUtil implements Accountable { 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 @@ public class SeriesScanUtil implements Accountable { } List<IPageReader> 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 @@ public class SeriesScanUtil implements Accountable { chunkMetaData.getVersion(), chunkMetaData.getOffsetOfChunkHeader(), iPageReader, - true)); + true, + tsFile)); } } else { for (int i = pageReaderList.size() - 1; i >= 0; i--) { @@ -729,7 +742,8 @@ public class SeriesScanUtil implements Accountable { chunkMetaData.getVersion(), chunkMetaData.getOffsetOfChunkHeader(), pageReaderList.get(i), - true)); + true, + tsFile)); } } } else { @@ -742,7 +756,8 @@ public class SeriesScanUtil implements Accountable { chunkMetaData.getVersion(), chunkMetaData.getOffsetOfChunkHeader(), pageReader, - false))); + false, + tsFile))); } if (LOGGER.isDebugEnabled()) { @@ -2083,6 +2098,7 @@ public class SeriesScanUtil implements Accountable { protected final boolean isSeq; protected final boolean isAligned; protected final boolean isMem; + protected final File tsFile; VersionPageReader( QueryContext context, @@ -2090,7 +2106,8 @@ public class SeriesScanUtil implements Accountable { 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 @@ public class SeriesScanUtil implements Accountable { || 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 class SeriesScanUtil implements Accountable { 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 4fb9e8a8cb5..7735ecae3e8 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.calc.execution.schedule.queue.IndexedBlockingQueue; 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 abstract class AbstractDriverThread extends Thread implements Closeable { 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 d16e602e3d2..2f2b0ba5db2 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.queryengine.plan.relational.planner.Symbol; 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; @@ -161,6 +162,12 @@ public class ExternalTsFileQueryResource { } } sealDeviceTaskPartitions(); + } catch (CorruptedTsFileException e) { + // Device collection runs during the planning phase, outside any DriverTask thread. + // CorruptedTsFileException is a RuntimeException and would crash the DataNode if not + // caught here. Re-wrap with initCause so the planning/query infrastructure can surface + // the error message to the client. + throw new RuntimeException(e.getMessage(), e); } } @@ -671,24 +678,45 @@ public class ExternalTsFileQueryResource { 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<Integer, LazyTsFileDeviceIterator> 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 +728,25 @@ public class ExternalTsFileQueryResource { deviceIteratorMap.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<Integer, LazyTsFileDeviceIterator> 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 +757,19 @@ public class ExternalTsFileQueryResource { private void collectCurrentDeviceOffsets() { List<ExternalTsFileDeviceQueryTask.DeviceOffset> deviceOffsets = new ArrayList<>(); for (Map.Entry<Integer, LazyTsFileDeviceIterator> 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 7c1e9d12635..5af55250285 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.IChunkReader; 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 class DiskAlignedChunkLoader implements IChunkLoader { 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 7d259a9e505..eb7abb7ed07 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.filter.basic.Filter; 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 class DiskChunkLoader implements IChunkLoader { 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/utils/ErrorHandlingUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java index 0a9216b8b28..ada3aac4412 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ErrorHandlingUtils.java @@ -27,6 +27,7 @@ import org.apache.iotdb.commons.exception.QuerySchemaFetchFailedException; import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils; import org.apache.iotdb.db.exception.BatchProcessException; +import org.apache.iotdb.db.exception.CorruptedTsFileException; import org.apache.iotdb.db.exception.QueryInBatchStatementException; import org.apache.iotdb.db.exception.StorageGroupNotReadyException; import org.apache.iotdb.db.exception.query.QueryTimeoutRuntimeException; @@ -186,6 +187,8 @@ public class ErrorHandlingUtils { : RpcUtils.getStatus(((IoTDBException) t).getErrorCode(), rootCause.getMessage()); } else if (t instanceof TsFileRuntimeException) { return RpcUtils.getStatus(TSStatusCode.TSFILE_PROCESSOR_ERROR, rootCause.getMessage()); + } else if (t instanceof CorruptedTsFileException) { + return RpcUtils.getStatus(TSStatusCode.TSFILE_PROCESSOR_ERROR, t.getMessage()); } else if (t instanceof SemanticException) { if (t.getCause() instanceof IoTDBException) { return RpcUtils.getStatus(
