diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadStorageTsFileWithCompactionIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadStorageTsFileWithCompactionIT.java new file mode 100644 index 0000000000000..bb8c79b921b18 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadStorageTsFileWithCompactionIT.java @@ -0,0 +1,269 @@ +/* + * 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.db.it; + +import org.apache.iotdb.it.env.MultiEnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.MultiClusterIT2DualTableManualBasic; +import org.apache.iotdb.itbase.env.BaseEnv; + +import org.apache.tsfile.common.constant.TsFileConstant; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.awaitility.Awaitility; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +@RunWith(IoTDBTestRunner.class) +@Category({MultiClusterIT2DualTableManualBasic.class}) +public class IoTDBLoadStorageTsFileWithCompactionIT { + + private static final String DATABASE = "load_storage_db"; + private static final String TABLE = "t1"; + private static final long COMPACTION_WAIT_SECONDS = 60L; + private static final String TABLE_MODEL_PROPERTY = "isTableModel"; + + private static BaseEnv producerEnv; + private static BaseEnv loaderEnv; + + @BeforeClass + public static void setUp() { + MultiEnvFactory.createEnv(2); + producerEnv = MultiEnvFactory.getEnv(0); + loaderEnv = MultiEnvFactory.getEnv(1); + + configure(producerEnv, false); + configure(loaderEnv, true); + + producerEnv.initClusterEnvironment(1, 1); + loaderEnv.initClusterEnvironment(1, 1); + } + + private static void configure(final BaseEnv env, final boolean enableCompaction) { + env.getConfig() + .getCommonConfig() + .setDatanodeMemoryProportion("3:3:1:1:1:0") + // The test starts one DataNode per cluster. Override the replication factors selected by + // the CI cluster mode (HighPerformanceMode uses three schema replicas) so that the + // database can be created with a single DataNode. + .setSchemaReplicationFactor(1) + .setDataReplicationFactor(1) + .setEnforceStrongPassword(false) + .setEnableSeqSpaceCompaction(enableCompaction) + .setEnableUnseqSpaceCompaction(enableCompaction) + .setEnableCrossSpaceCompaction(enableCompaction) + .setInnerCompactionCandidateFileNum(2); + // Make the test deterministic: the producer creates two files with compaction disabled, + // then enables it and waits for the scheduler to merge them. + env.getConfig().getDataNodeConfig().setCompactionScheduleInterval(1000); + } + + @AfterClass + public static void tearDown() { + if (loaderEnv != null) { + loaderEnv.cleanClusterEnvironment(); + } + if (producerEnv != null) { + producerEnv.cleanClusterEnvironment(); + } + } + + @Test + public void testLoadStorageTsFileAfterCompaction() throws Exception { + final Path tempDir = Files.createTempDirectory("load-storage-tsfile"); + try { + final long fileCountBeforeCompaction = createTableFilesOnProducer(); + Assert.assertTrue( + "The producer should create two TsFiles before compaction, but got " + + fileCountBeforeCompaction, + fileCountBeforeCompaction >= 2); + assertAllTsFilesHaveTableSchema(producerEnv); + + enableSeqSpaceCompaction(producerEnv); + Awaitility.await() + .pollInterval(200, TimeUnit.MILLISECONDS) + .atMost(COMPACTION_WAIT_SECONDS, TimeUnit.SECONDS) + .until(() -> countTsFiles(producerEnv) == 1); + + final Path sourceTsFile = findSingleTsFile(producerEnv); + assertTableSchemaRegistered(sourceTsFile); + + final Path copiedTsFile = tempDir.resolve(sourceTsFile.getFileName().toString()); + Files.copy(sourceTsFile, copiedTsFile, StandardCopyOption.REPLACE_EXISTING); + + loadIntoReceiver(copiedTsFile); + assertLoadedRowCount(2); + } finally { + deleteRecursively(tempDir); + } + } + + private static long createTableFilesOnProducer() throws Exception { + try (final Connection connection = producerEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE IF NOT EXISTS " + DATABASE); + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE IF NOT EXISTS " + TABLE + " (tag1 STRING TAG, s1 INT32 FIELD)"); + + // Use the same tag and timestamps near the current time so that both files belong to the + // same data region; otherwise each file is compacted independently. + final long baseTime = System.currentTimeMillis(); + for (int i = 0; i < 2; i++) { + statement.execute( + String.format( + "INSERT INTO %s(time, tag1, s1) VALUES (%d, 'tag', %d)", TABLE, baseTime + i, i)); + statement.execute("FLUSH"); + } + } + + Awaitility.await() + .pollInterval(200, TimeUnit.MILLISECONDS) + .atMost(COMPACTION_WAIT_SECONDS, TimeUnit.SECONDS) + .until(() -> countTsFiles(producerEnv) >= 2); + return countTsFiles(producerEnv); + } + + private static void enableSeqSpaceCompaction(final BaseEnv env) throws Exception { + try (final Connection connection = env.getConnection(BaseEnv.TABLE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute("SET CONFIGURATION enable_seq_space_compaction='true'"); + } + } + + private static void loadIntoReceiver(final Path tsFile) throws Exception { + try (final Connection connection = loaderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE IF NOT EXISTS " + DATABASE); + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE IF NOT EXISTS " + TABLE + " (tag1 STRING TAG, s1 INT32 FIELD)"); + statement.execute(String.format("LOAD '%s'", tsFile.toAbsolutePath())); + } + } + + private static void assertLoadedRowCount(final long expected) throws Exception { + try (final Connection connection = loaderEnv.getConnection(BaseEnv.TABLE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + try (final ResultSet resultSet = statement.executeQuery("SELECT count(*) FROM " + TABLE)) { + Assert.assertTrue("LOAD result set is empty", resultSet.next()); + Assert.assertEquals(expected, resultSet.getLong(1)); + } + } + } + + private static void assertTableSchemaRegistered(final Path tsFile) throws IOException { + try (final TsFileSequenceReader reader = + new TsFileSequenceReader(tsFile.toAbsolutePath().toString())) { + final Map tableSchemaMap = reader.getTableSchemaMap(); + Assert.assertNotNull("TsFile has no table schema map", tableSchemaMap); + Assert.assertTrue( + "TsFile has no schema entry for table " + TABLE + ": " + tableSchemaMap, + tableSchemaMap.containsKey(TABLE)); + Assert.assertFalse( + "The storage path must not synthesize a TIME column: " + tableSchemaMap.get(TABLE), + tableSchemaMap.get(TABLE).getColumnTypes().contains(ColumnCategory.TIME)); + Assert.assertTrue( + "A table-model TsFile must carry the " + TABLE_MODEL_PROPERTY + " property", + reader.readFileMetadata().getTsFileProperties().containsKey(TABLE_MODEL_PROPERTY)); + } + } + + private static void assertAllTsFilesHaveTableSchema(final BaseEnv env) throws IOException { + final Path dataPath = Paths.get(env.getDataNodeWrapper(0).getDataPath()); + try (final Stream paths = Files.walk(dataPath)) { + final List tsFiles = + paths + .filter(Files::isRegularFile) + .filter(path -> path.getFileName().toString().endsWith(TsFileConstant.TSFILE_SUFFIX)) + .collect(Collectors.toList()); + Assert.assertFalse("No TsFile was produced", tsFiles.isEmpty()); + for (final Path tsFile : tsFiles) { + assertTableSchemaRegistered(tsFile); + } + } + } + + private static long countTsFiles(final BaseEnv env) throws IOException { + final Path dataPath = Paths.get(env.getDataNodeWrapper(0).getDataPath()); + if (!Files.exists(dataPath)) { + return 0; + } + try (final Stream paths = Files.walk(dataPath)) { + return paths + .filter(Files::isRegularFile) + .filter(path -> path.getFileName().toString().endsWith(TsFileConstant.TSFILE_SUFFIX)) + .count(); + } + } + + private static Path findSingleTsFile(final BaseEnv env) throws IOException { + final Path dataPath = Paths.get(env.getDataNodeWrapper(0).getDataPath()); + try (final Stream paths = Files.walk(dataPath)) { + final List tsFiles = + paths + .filter(Files::isRegularFile) + .filter(path -> path.getFileName().toString().endsWith(TsFileConstant.TSFILE_SUFFIX)) + .collect(Collectors.toList()); + Assert.assertEquals( + "Expected exactly one compacted TsFile, but got " + tsFiles, 1, tsFiles.size()); + return tsFiles.get(0); + } + } + + private static void deleteRecursively(final Path path) throws IOException { + if (!Files.exists(path)) { + return; + } + try (final Stream paths = Files.walk(path)) { + paths + .sorted((left, right) -> right.getNameCount() - left.getNameCount()) + .forEach( + current -> { + try { + Files.deleteIfExists(current); + } catch (final IOException e) { + throw new RuntimeException(e); + } + }); + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTableSchemaCompletionIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTableSchemaCompletionIT.java new file mode 100644 index 0000000000000..57bb21d42be1b --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTableSchemaCompletionIT.java @@ -0,0 +1,265 @@ +/* + * 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.db.it; + +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.TableClusterIT; +import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; +import org.apache.iotdb.itbase.env.BaseEnv; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.StringArrayDeviceID; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.stream.Stream; + +/** + * Tests the LOAD behavior of table-model TsFiles whose TableSchema entry does not describe all the + * columns of the table. + * + *

The storage path writes a time-only entry for a table that has no column other than time, and + * a file written before the table-model file property existed may carry no entry at all. LOAD must + * complete such a schema from the table in IoTDB instead of failing the whole file, and must still + * create the table with every column when the file does describe them. + */ +@RunWith(IoTDBTestRunner.class) +@Category({TableLocalStandaloneIT.class, TableClusterIT.class}) +public class IoTDBLoadTableSchemaCompletionIT { + + private static final String DATABASE = "load_schema_completion_db"; + private static final String TIME_ONLY_TABLE = "t_time_only"; + private static final String PIPE_TABLE = "t_pipe"; + private static final String AUTO_CREATE_TABLE = "t_auto_create"; + private static final String TAG_VALUE = "tagA"; + private static final int ROW_COUNT = 3; + + /** The file property written by every table-model writer (flush, compaction, recovery, pipe). */ + private static final String TABLE_MODEL_PROPERTY = "isTableModel"; + + /** What the TableSchema entry of the generated TsFile contains. */ + private enum SchemaEntry { + /** Every column of the table, which is what a flush or a compaction writes. */ + FULL, + /** Only the time column, which is what a table without another column produces. */ + TIME_ONLY, + /** No entry at all, which a file written before the file property existed can carry. */ + ABSENT + } + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv().initClusterEnvironment(); + try (final Connection connection = + EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE IF NOT EXISTS " + DATABASE); + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE IF NOT EXISTS " + TIME_ONLY_TABLE + " (tag1 STRING TAG, s1 INT32 FIELD)"); + statement.execute( + "CREATE TABLE IF NOT EXISTS " + PIPE_TABLE + " (tag1 STRING TAG, s1 INT32 FIELD)"); + // This one must stay absent so that LOAD has to create it. + statement.execute("DROP TABLE IF EXISTS " + AUTO_CREATE_TABLE); + } + } + + @AfterClass + public static void tearDown() { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testLoadTsFileWithTimeOnlyTableSchema() throws Exception { + // The schema entry keeps only the time column but the chunks still hold s1. + loadAndVerify(TIME_ONLY_TABLE, SchemaEntry.TIME_ONLY); + } + + @Test + public void testLoadTsFileWithoutTableSchemaEntry() throws Exception { + // A legacy table-model file: the isTableModel property is set but no schema entry exists. + loadAndVerify(PIPE_TABLE, SchemaEntry.ABSENT); + } + + @Test + public void testLoadTsFileCreatesTableWithAllColumns() throws Exception { + // The table does not exist and the file describes every column, so LOAD has to create the + // table with all of them, including the tag column that is only carried by the device id. + loadAndVerify(AUTO_CREATE_TABLE, SchemaEntry.FULL); + } + + private static void loadAndVerify(final String tableName, final SchemaEntry schemaEntry) + throws Exception { + final Path tempDir = Files.createTempDirectory("load-schema-completion"); + try { + final Path tsFile = writeTableModelTsFile(tempDir, tableName, schemaEntry); + assertTsFileContent(tsFile, tableName, schemaEntry); + + try (final Connection connection = + EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute(String.format("LOAD '%s'", tsFile.toAbsolutePath())); + + try (final ResultSet resultSet = + statement.executeQuery("SELECT tag1, s1 FROM " + tableName + " ORDER BY time")) { + int expectedValue = 0; + while (resultSet.next()) { + Assert.assertEquals(TAG_VALUE, resultSet.getString(1)); + Assert.assertEquals(expectedValue, resultSet.getInt(2)); + expectedValue++; + } + Assert.assertEquals(ROW_COUNT, expectedValue); + } + } + } finally { + deleteRecursively(tempDir); + } + } + + private static Path writeTableModelTsFile( + final Path dir, final String tableName, final SchemaEntry schemaEntry) throws Exception { + final Path tsFile = dir.resolve(tableName + ".tsfile"); + final List measurementSchemas = + Arrays.asList( + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("s1", TSDataType.INT32)); + try (final TsFileIOWriter writer = new TsFileIOWriter(tsFile.toFile())) { + switch (schemaEntry) { + case FULL: + writer + .getSchema() + .registerTableSchema( + new TableSchema( + tableName, + measurementSchemas, + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD))); + break; + case TIME_ONLY: + // Only the time column is kept in the schema entry, which is what the storage path + // writes when the columns of the table are no longer available. + writer.getSchema().registerTableSchema(new TableSchema(tableName)); + break; + case ABSENT: + default: + break; + } + writer.startChunkGroup(new StringArrayDeviceID(tableName, TAG_VALUE)); + // The tag column is carried by the device id, so only s1 is written into the chunk group. + final AlignedChunkWriterImpl chunkWriter = + new AlignedChunkWriterImpl(measurementSchemas.subList(1, measurementSchemas.size())); + for (int i = 0; i < ROW_COUNT; i++) { + final long time = 100L + i; + chunkWriter.getTimeChunkWriter().write(time); + chunkWriter.getValueChunkWriterByIndex(0).write(time, i, false); + } + chunkWriter.writeToFileWriter(writer); + writer.endChunkGroup(); + writer.addTsFileProperty(TABLE_MODEL_PROPERTY, "true".getBytes(StandardCharsets.UTF_8)); + writer.endFile(); + } + return tsFile; + } + + private static void assertTsFileContent( + final Path tsFile, final String tableName, final SchemaEntry schemaEntry) throws Exception { + try (final TsFileSequenceReader reader = + new TsFileSequenceReader(tsFile.toAbsolutePath().toString())) { + Assert.assertTrue( + "The TsFile must be marked as a table-model file", + reader.readFileMetadata().getTsFileProperties().containsKey(TABLE_MODEL_PROPERTY)); + + final Map tableSchemaMap = reader.getTableSchemaMap(); + Assert.assertNotNull(tableSchemaMap); + switch (schemaEntry) { + case FULL: + Assert.assertTrue( + "TsFile has no schema entry for table " + tableName + ": " + tableSchemaMap, + tableSchemaMap.containsKey(tableName)); + Assert.assertEquals( + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD), + tableSchemaMap.get(tableName).getColumnTypes()); + break; + case TIME_ONLY: + Assert.assertTrue( + "TsFile has no schema entry for table " + tableName + ": " + tableSchemaMap, + tableSchemaMap.containsKey(tableName)); + Assert.assertTrue( + "The schema entry must not contain any column other than time: " + + tableSchemaMap.get(tableName), + tableSchemaMap.get(tableName).getColumnSchemas().isEmpty()); + break; + case ABSENT: + default: + Assert.assertTrue( + "The TsFile must not contain any schema entry: " + tableSchemaMap, + tableSchemaMap.isEmpty()); + break; + } + Assert.assertTrue( + "The TsFile must contain device " + tableName + "." + TAG_VALUE, + reader.getAllDevices().stream() + .anyMatch( + device -> + Arrays.equals(device.getSegments(), new String[] {tableName, TAG_VALUE}))); + } + } + + private static void deleteRecursively(final Path path) throws IOException { + if (!Files.exists(path)) { + return; + } + try (final Stream stream = Files.walk(path)) { + stream + .sorted(Comparator.reverseOrder()) + .forEach( + file -> { + try { + Files.deleteIfExists(file); + } catch (final Exception e) { + // Ignore cleanup failures, the temporary directory is not part of the assertion. + } + }); + } + } +} 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 c326a7d0fd24a..9099bd01d17fa 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 @@ -2747,6 +2747,12 @@ public final class DataNodeQueryMessages { "Failed to check if device {} is deleted by mods. Will see it as not deleted."; public static final String COLUMN_ARG_IN_TABLE_ARG_IS_NOT_FOUND_IN_IOTDB_WHILE_LOADING_TSFILE = "Column {} in table {} is not found in IoTDB while loading TsFile."; + public static final String EXCEPTION_THE_SCHEMA_OF_TABLE_ARG_IS_MISSING_IN_THE_TSFILE_AND_THE_TABLE_DOES_NOT_EXIST_IN_DATABASE_ARG_SO_THE_FILE_CANNOT_BE_LOADED_D910A1AA = + "The schema of table %s is missing in the TsFile and the table does not exist in database %s, so the file cannot be loaded."; + public static final String LOG_TABLE_ARG_HAS_NO_SCHEMA_ENTRY_IN_THE_TSFILE_WILL_USE_THE_SCHEMA_OF_THE_TABLE_IN_IOTDB_F7ED883F = + "Table {} has no schema entry in the TsFile, will use the schema of the table in IoTDB."; + public static final String LOG_TABLE_ARG_HAS_NO_COLUMN_OTHER_THAN_TIME_IN_ITS_SCHEMA_ENTRY_IN_THE_TSFILE_WILL_USE_THE_SCHEMA_OF_THE_TABLE_IN_IOTDB_EB6C0BE1 = + "Table {} has no column other than time in its schema entry in the TsFile, will use the schema of the table in IoTDB."; public static final String DEVICE_ARG_IS_NOT_IN_THE_TSFILEDEVICE2ISALIGNED_CACHE_ARG = "Device {} is not in the tsFileDevice2IsAligned cache {}."; public static final String LOADTSFILEANALYZER_CURRENT_DATANODE_IS_READ_ONLY_WILL_TRY_TO_CONVERT_TO_TABLETS_AND = 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 a406b246c4ea2..d8783e4e6210a 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 @@ -3168,6 +3168,12 @@ public final class DataNodeQueryMessages { public static final String COLUMN_ARG_IN_TABLE_ARG_IS_NOT_FOUND_IN_IOTDB_WHILE_LOADING_TSFILE = "加载 TsFile 时，表 {} 中的列 {} 在 IoTDB 中未找到。"; + public static final String EXCEPTION_THE_SCHEMA_OF_TABLE_ARG_IS_MISSING_IN_THE_TSFILE_AND_THE_TABLE_DOES_NOT_EXIST_IN_DATABASE_ARG_SO_THE_FILE_CANNOT_BE_LOADED_D910A1AA = + "TsFile 中缺少表 %s 的 schema，且数据库 %s 中不存在该表，无法加载该文件。"; + public static final String LOG_TABLE_ARG_HAS_NO_SCHEMA_ENTRY_IN_THE_TSFILE_WILL_USE_THE_SCHEMA_OF_THE_TABLE_IN_IOTDB_F7ED883F = + "TsFile 中缺少表 {} 的 schema，将使用 IoTDB 中该表的 schema。"; + public static final String LOG_TABLE_ARG_HAS_NO_COLUMN_OTHER_THAN_TIME_IN_ITS_SCHEMA_ENTRY_IN_THE_TSFILE_WILL_USE_THE_SCHEMA_OF_THE_TABLE_IN_IOTDB_EB6C0BE1 = + "TsFile 中表 {} 的 schema 只有 Time 列，将使用 IoTDB 中该表的 schema。"; public static final String DEVICE_ARG_IS_NOT_IN_THE_TSFILEDEVICE2ISALIGNED_CACHE_ARG = "设备 {} 不在 tsFileDevice2IsAligned 缓存 {} 中。"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java index ad11596a8d566..03096f93b0374 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/builder/PipeTableModelTsFileBuilderV2.java @@ -28,6 +28,7 @@ import org.apache.iotdb.db.storageengine.dataregion.flush.MemTableFlushTask; import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; import org.apache.iotdb.db.storageengine.dataregion.memtable.PrimitiveMemTable; +import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileTableSchemaRegistrationUtils; import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.tsfile.encrypt.EncryptUtils; @@ -339,6 +340,9 @@ private void writeTabletsIntoOneFile( new MemTableFlushTask(memTable, writer, null, null, true); memTableFlushTask.syncFlushMemTable(); + // The marker is written once, after the last tablet is flushed, so building the file pays + // nothing per tablet. + TsFileTableSchemaRegistrationUtils.markTableModel(writer); writer.endFile(); } } 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 2e6a978937d13..beeea9b56c145 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 @@ -48,6 +48,7 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils; +import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileTableSchemaRegistrationUtils; import org.apache.iotdb.db.storageengine.load.active.ActiveLoadPathHelper; import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter; import org.apache.iotdb.db.storageengine.load.converter.PipeTsFileConversionTaskManager; @@ -426,7 +427,7 @@ private void analyzeSingleTsFile(final File tsFile, int i) throws Exception { try (final TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) { // check whether the tsfile is tree-model or not final Map tableSchemaMap = reader.getTableSchemaMap(); - final boolean isTableModelFile = Objects.nonNull(tableSchemaMap) && !tableSchemaMap.isEmpty(); + final boolean isTableModelFile = detectTableModelTsFile(reader, tableSchemaMap); if (isGeneratedByPipe) { LOGGER.debug( DataNodeQueryMessages.TSFILE_ARG_IS_A_ARG_MODEL_FILE, @@ -506,6 +507,19 @@ && handleSingleMiniFile(i)) { } } + /** + * Whether the TsFile is a table-model file. The explicit file property is authoritative, so a + * table-model file whose TableSchema map is empty is not mistaken for a tree-model file. Files + * written before the property existed fall back to the non-empty TableSchema map. + */ + static boolean detectTableModelTsFile( + final TsFileSequenceReader reader, final Map tableSchemaMap) + throws IOException { + // Keep the cheap map check first so that a normal file never reads the metadata again. + return (Objects.nonNull(tableSchemaMap) && !tableSchemaMap.isEmpty()) + || TsFileTableSchemaRegistrationUtils.isTableModel(reader.readFileMetadata()); + } + private void deleteFile(final File file) { try { RetryUtils.retryOnException( @@ -933,7 +947,7 @@ private void getFileModelInfoBeforeTabletConversion() throws IOException { try (final TsFileSequenceReader reader = new TsFileSequenceReader(tsFiles.get(i).getAbsolutePath(), true)) { final Map tableSchemaMap = reader.getTableSchemaMap(); - isTableModelTsFile.set(i, Objects.nonNull(tableSchemaMap) && !tableSchemaMap.isEmpty()); + isTableModelTsFile.set(i, detectTableModelTsFile(reader, tableSchemaMap)); isTableModelTsFileReliableIndex = i; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java index 63a8adcaea955..00887ad8dc48f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileTableSchemaCache.java @@ -25,6 +25,8 @@ import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.TableSchema; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; import org.apache.iotdb.db.auth.AuthorityChecker; @@ -97,6 +99,10 @@ public class LoadTsFileTableSchemaCache { private String database; private boolean needToCreateDatabase; private Map tableSchemaMap; + + /** Tables whose schema has already been created and verified for the file being analyzed. */ + private final Set verifiedTables = new HashSet<>(); + private final Metadata metadata; private final MPPQueryContext context; private final boolean shouldVerifyDataType; @@ -141,6 +147,7 @@ public void setDatabase(final String database) { public void setTableSchemaMap( final Map tableSchemaMap) { this.tableSchemaMap = tableSchemaMap; + this.verifiedTables.clear(); } public void autoCreateAndVerify(final IDeviceID device) throws LoadAnalyzeException { @@ -296,10 +303,45 @@ private static Object[] truncateNullSuffixesOfDeviceIdSegments(Object[] segments public void createTableAndDatabaseIfNecessary(final String tableName) throws LoadAnalyzeException { - final org.apache.tsfile.file.metadata.TableSchema schema = tableSchemaMap.remove(tableName); - if (Objects.isNull(schema)) { + final Map fileTableSchemaMap = + this.tableSchemaMap; + org.apache.tsfile.file.metadata.TableSchema schema = + Objects.isNull(fileTableSchemaMap) ? null : fileTableSchemaMap.remove(tableName); + if (!verifiedTables.add(tableName)) { + // Another device of this table has already triggered the creation and verification. return; } + if (Objects.isNull(schema)) { + // The TsFile has no schema entry for this table. Complete it from the table in IoTDB so that + // the table is still created and verified and the tag column mapping stays correct, instead + // of silently skipping the whole check. + schema = resolveTableSchemaFromCache(tableName); + if (Objects.isNull(schema)) { + throw new LoadAnalyzeException( + String.format( + DataNodeQueryMessages + .EXCEPTION_THE_SCHEMA_OF_TABLE_ARG_IS_MISSING_IN_THE_TSFILE_AND_THE_TABLE_DOES_NOT_EXIST_IN_DATABASE_ARG_SO_THE_FILE_CANNOT_BE_LOADED_D910A1AA, + tableName, + database)); + } + LOGGER.warn( + DataNodeQueryMessages + .LOG_TABLE_ARG_HAS_NO_SCHEMA_ENTRY_IN_THE_TSFILE_WILL_USE_THE_SCHEMA_OF_THE_TABLE_IN_IOTDB_F7ED883F, + tableName); + } else if (isTimeOnlyTableSchema(schema)) { + // The schema entry keeps no column other than time (e.g. it was completed after the table + // had been dropped or a pipe-built file carries only the table name). Such a schema cannot + // pass the header validation, complete it from the table in IoTDB when the table exists. + final org.apache.tsfile.file.metadata.TableSchema cachedSchema = + resolveTableSchemaFromCache(tableName); + if (Objects.nonNull(cachedSchema)) { + schema = cachedSchema; + LOGGER.warn( + DataNodeQueryMessages + .LOG_TABLE_ARG_HAS_NO_COLUMN_OTHER_THAN_TIME_IN_ITS_SCHEMA_ENTRY_IN_THE_TSFILE_WILL_USE_THE_SCHEMA_OF_THE_TABLE_IN_IOTDB_EB6C0BE1, + tableName); + } + } // Check on creation, do not auto-create tables or database that cannot be inserted AuthorityChecker.getAccessControl() @@ -333,6 +375,19 @@ public boolean isNeedDecode4DifferentTimeColumn() { return needDecode4DifferentTimeColumn.get(); } + private org.apache.tsfile.file.metadata.TableSchema resolveTableSchemaFromCache( + final String tableName) { + final TsTable table = DataNodeTableCache.getInstance().getTable(database, tableName, false); + return Objects.isNull(table) + ? null + : TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table); + } + + private static boolean isTimeOnlyTableSchema( + final org.apache.tsfile.file.metadata.TableSchema schema) { + return Objects.isNull(schema.getColumnSchemas()) || schema.getColumnSchemas().isEmpty(); + } + private void autoCreateTableDatabaseIfAbsent(final String database) throws LoadAnalyzeException { validateDatabaseName(database); if (DataNodeTableCache.getInstance().isDatabaseExist(database)) { @@ -487,6 +542,7 @@ public void close() { clearDevices(); clearTagColumnMapper(); clearModificationsAndTimeIndex(); + verifiedTables.clear(); block.close(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 52000fe97fce3..07f1dca081021 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -1020,7 +1020,8 @@ private void checkTsFileTime(File tsFile, long currentTime) throws DataRegionExc private WALRecoverListener recoverUnsealedTsFile( TsFileResource unsealedTsFile, DataRegionRecoveryContext context, boolean isSeq) { UnsealedTsFileRecoverPerformer recoverPerformer = - new UnsealedTsFileRecoverPerformer(unsealedTsFile, isSeq, context.recoverPerformers::add); + new UnsealedTsFileRecoverPerformer( + unsealedTsFile, isSeq, isTableModel, context.recoverPerformers::add); // remember to close UnsealedTsFileRecoverPerformer return WALRecoverManager.getInstance().addRecoverPerformer(recoverPerformer); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java index 82bdbd3e6496d..1c3a6934d4d7e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.utils.MetadataUtils; +import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.StorageEngineException; @@ -76,6 +77,7 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer { private Schema schema = null; private EncryptParameter firstEncryptParameter; + private boolean tableModel; @TestOnly public ReadChunkCompactionPerformer(List sourceFiles, TsFileResource targetFile) { @@ -138,6 +140,7 @@ public void perform() StorageEngineException, PageException { try (MultiTsFileDeviceIterator deviceIterator = new MultiTsFileDeviceIterator(seqFiles)) { + tableModel = PathUtils.isTableModelDatabase(deviceIterator.getDatabaseName()); schema = CompactionTableSchemaCollector.collectSchema( seqFiles, @@ -218,6 +221,7 @@ private void useNewWriter() throws IOException { memoryBudgetForFileWriter, CompactionType.INNER_SEQ_COMPACTION, firstEncryptParameter); + currentWriter.setTableModel(tableModel); summary.recordTargetTsFileTableSizeMap( targetResources.get(currentTargetFileIndex), currentWriter.getTableSizeMap()); currentWriter.setSchema(CompactionTableSchemaCollector.copySchema(schema)); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/io/CompactionTsFileWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/io/CompactionTsFileWriter.java index 53754a8a2ac0f..45df5195f2d0e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/io/CompactionTsFileWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/io/CompactionTsFileWriter.java @@ -24,6 +24,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionTaskManager; import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.constant.CompactionIoDataType; import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.constant.CompactionType; +import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileTableSchemaRegistrationUtils; import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.tsfile.encrypt.EncryptParameter; @@ -55,6 +56,7 @@ public class CompactionTsFileWriter extends TsFileIOWriter { private volatile boolean isWritingAligned = false; private boolean isEmptyTargetFile = true; private IDeviceID currentDeviceId; + private boolean tableModel; private EncryptParameter firstEncryptParameter; @@ -91,6 +93,10 @@ public static CompactionTsFileWriter create( } } + public void setTableModel(final boolean tableModel) { + this.tableModel = tableModel; + } + public void markStartingWritingAligned() { isWritingAligned = true; } @@ -173,6 +179,11 @@ public void endChunkGroup() throws IOException { @Override public void endFile() throws IOException { + // Every table-model TsFile carries the marker, whether or not its TableSchema map keeps an + // entry, so readers do not have to infer the model from the schema map. + if (tableModel) { + TsFileTableSchemaRegistrationUtils.markTableModel(this); + } removeUnusedTableSchema(); long beforeSize = this.getPos(); super.endFile(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 6ca52288e8249..aeed7070d4f07 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -68,6 +68,7 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndexCacheRecorder; import org.apache.iotdb.db.storageengine.dataregion.utils.SharedTimeDataBuffer; +import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileTableSchemaRegistrationUtils; import org.apache.iotdb.db.storageengine.dataregion.utils.tableDiskUsageIndex.TableDiskUsageIndex; import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; import org.apache.iotdb.db.storageengine.dataregion.wal.node.IWALNode; @@ -159,6 +160,14 @@ public class TsFileProcessor { private final Map registeredTableSchemaVersions = new ConcurrentHashMap<>(); + /** + * Whether this processor writes a table-model TsFile. Such a file is marked with a single {@link + * TsFileTableSchemaRegistrationUtils#TABLE_MODEL_PROPERTY} property at {@link #endFile()}, so the + * closed file stays recognizable as a table-model TsFile without any per-table schema lookup on + * the write path. + */ + private final boolean tableModelTsFile; + /** Tsfile resource for index this tsfile. */ private final TsFileResource tsFileResource; @@ -246,6 +255,7 @@ public TsFileProcessor( this.writer = EncryptDBUtils.createRestorableWriterForNewFile( tsfile, dataRegionInfo.getDataRegion().getDatabaseName()); + this.tableModelTsFile = dataRegionInfo.getDataRegion().isTableModel(); this.updateLatestFlushTimeCallback = updateLatestFlushTimeCallback; this.walNode = WALManager.getInstance() @@ -269,6 +279,7 @@ public TsFileProcessor( this.tsFileResource = tsFileResource; this.dataRegionInfo = dataRegionInfo; this.writer = writer; + this.tableModelTsFile = dataRegionInfo.getDataRegion().isTableModel(); this.updateLatestFlushTimeCallback = updateLatestFlushTimeCallback; this.sequence = sequence; this.walNode = @@ -2233,6 +2244,12 @@ private void endFile() throws IOException, TsFileProcessorException { if (logger.isDebugEnabled()) { logger.debug(StorageEngineMessages.START_TO_END_FILE, tsFileResource); } + // Mark the file as a table-model TsFile. Written once here, when the file is sealed: it is a + // single entry of the in-memory file property map serialized by endFile(), never a per-device + // or per-chunk operation. + if (tableModelTsFile) { + TsFileTableSchemaRegistrationUtils.markTableModel(writer); + } writer.endFile(); // Listen after "endFile" to avoid unnecessary waiting for tsFile close diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtils.java new file mode 100644 index 0000000000000..78a4165efed51 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtils.java @@ -0,0 +1,110 @@ +/* + * 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.dataregion.utils; + +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; + +import org.apache.tsfile.file.metadata.TsFileMetadata; +import org.apache.tsfile.write.writer.TsFileIOWriter; + +import java.nio.charset.StandardCharsets; +import java.util.Objects; + +/** + * Keeps a table-model TsFile identifiable and self-describing. + * + *

Every table-model TsFile carries a single {@value #TABLE_MODEL_PROPERTY} file property. It is + * a constant-time marker that tells readers the file belongs to the table model, so a file whose + * TableSchema map is empty is not mistaken for a tree-model TsFile. + * + *

The TableSchema map keeps one entry per table that writes data into the file. This class only + * changes map membership: it never adds, removes, or rewrites columns in an existing TableSchema. + * In particular it does not synthesize an entry for a table that no longer exists, the file stays + * recognizable through the file property instead. + */ +public final class TsFileTableSchemaRegistrationUtils { + + /** File property written by every table-model TsFile; its value is {@code true}. */ + public static final String TABLE_MODEL_PROPERTY = "isTableModel"; + + private static final byte[] TABLE_MODEL_PROPERTY_VALUE = "true".getBytes(StandardCharsets.UTF_8); + + private TsFileTableSchemaRegistrationUtils() {} + + /** + * Records that the file being written belongs to the table model. + * + *

Call it once per file, when the file is sealed, that is, immediately before {@link + * TsFileIOWriter#endFile()}. The marker is a single entry of the in-memory file property map and + * is serialized together with the file metadata by {@code endFile()}, so it costs nothing on the + * write path and must never be called per record, per tablet or per chunk group. + */ + public static void markTableModel(final TsFileIOWriter writer) { + if (writer == null) { + return; + } + writer.addTsFileProperty(TABLE_MODEL_PROPERTY, TABLE_MODEL_PROPERTY_VALUE); + } + + /** + * Whether the given file metadata says that the TsFile was written as a table-model file. This + * does not look at the TableSchema map, so it also recognizes a table-model file whose schema map + * is empty. + */ + public static boolean isTableModel(final TsFileMetadata metadata) { + return metadata != null + && metadata.getTsFileProperties() != null + && metadata.getTsFileProperties().containsKey(TABLE_MODEL_PROPERTY); + } + + /** + * Registers the TableSchema of {@code tableName} when the writer does not have it yet. + * + *

This is called while an unsealed table-model file is recovered: the TableSchema of a table + * whose data is replayed from the WAL has to be registered again, otherwise the closed file would + * carry the chunks of the table but no description of them. + * + *

A table that no longer exists is skipped instead of being written with an empty schema: the + * storage path does not describe tables it cannot read, and the file property already tells + * readers that the file is a table-model one. + * + *

The caller invokes this only for a table-model DataRegion, so {@code databaseName} is used + * here solely as a cache key and is never inspected to guess the model. Inside this method the + * table cache is the only source of truth: a table that no longer exists has no entry and is + * skipped, and a tree-model database never owns a {@link TsTable} at all. + */ + public static void registerTableSchemaIfAbsent( + final TsFileIOWriter writer, final String databaseName, final String tableName) { + if (Objects.isNull(writer) + || Objects.isNull(tableName) + || writer.getSchema().getTableSchemaMap().containsKey(tableName)) { + return; + } + final TsTable table = DataNodeTableCache.getInstance().getTable(databaseName, tableName, false); + if (Objects.isNull(table)) { + return; + } + writer + .getSchema() + .registerTableSchema(TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table)); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java index cdfdc13dba669..56e0f49dbc02e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformer.java @@ -20,8 +20,6 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.recover.file; import org.apache.iotdb.commons.path.MeasurementPath; -import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; -import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.db.exception.DataRegionException; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; @@ -29,7 +27,6 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; -import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.db.service.metrics.DataNodeExceptionMetrics; import org.apache.iotdb.db.storageengine.dataregion.flush.CompressionRatio; import org.apache.iotdb.db.storageengine.dataregion.flush.MemTableFlushTask; @@ -43,6 +40,7 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndexCacheRecorder; +import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileTableSchemaRegistrationUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALRecoverException; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALRecoverListener; @@ -91,11 +89,12 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform public UnsealedTsFileRecoverPerformer( TsFileResource tsFileResource, boolean sequence, + boolean isTableModel, Consumer callbackAfterUnsealedTsFileRecovered) { super(tsFileResource); this.databaseName = tsFileResource.getDatabaseName(); this.dataRegionId = tsFileResource.getDataRegionId(); - this.isTableModel = PathUtils.isTableModelDatabase(databaseName); + this.isTableModel = isTableModel; this.sequence = sequence; this.callbackAfterUnsealedTsFileRecovered = callbackAfterUnsealedTsFileRecovered; this.walRedoer = new TsFilePlanRedoer(tsFileResource); @@ -253,16 +252,13 @@ public void redoLog(WALEntry walEntry) { } private void registerToTsFile(String tableName) { - if (tableName != null) { - writer - .getSchema() - .getTableSchemaMap() - .computeIfAbsent( - tableName, - t -> - TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute( - DataNodeTableCache.getInstance().getTable(databaseName, t))); + // Only a table-model DataRegion owns TsTables, so only its recovered file may carry a + // TableSchema entry. The model comes from the owning DataRegion instead of being derived from + // the database name here. + if (!isTableModel) { + return; } + TsFileTableSchemaRegistrationUtils.registerTableSchemaIfAbsent(writer, databaseName, tableName); } /** @@ -326,6 +322,9 @@ public void endRecovery() throws WALRecoverException { // if we put following codes in the 'if' clause above, this file can be continued writing // into it // currently, we close this file anyway + if (isTableModel) { + TsFileTableSchemaRegistrationUtils.markTableModel(writer); + } writer.endFile(); tsFileResource.serialize(); FileTimeIndexCacheRecorder.getInstance().logFileTimeIndex(tsFileResource); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java index 8b3217bad2f98..04e370469bb11 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java @@ -735,7 +735,10 @@ public void testUncheckedWriteAfterWalManagerStopLosesCheckpoint() throws Except recoverManager.clear(); final UnsealedTsFileRecoverPerformer performer = new UnsealedTsFileRecoverPerformer( - new TsFileResource(lateFile), true, recovered -> assertFalse(recovered.canWrite())); + new TsFileResource(lateFile), + true, + false, + recovered -> assertFalse(recovered.canWrite())); recoverManager.addRecoverPerformer(performer); recoverManager.setAllDataRegionScannedLatch(new ExceptionalCountDownLatch(0)); recoverManager.recover(); @@ -795,7 +798,10 @@ public void testAsyncWriteAfterTsFileFlushIsRecoveredWhenWalCloses() throws Exce recoverManager.clear(); final UnsealedTsFileRecoverPerformer performer = new UnsealedTsFileRecoverPerformer( - new TsFileResource(lateFile), true, recovered -> assertFalse(recovered.canWrite())); + new TsFileResource(lateFile), + true, + false, + recovered -> assertFalse(recovered.canWrite())); recoverManager.addRecoverPerformer(performer); recoverManager.setAllDataRegionScannedLatch(new ExceptionalCountDownLatch(0)); recoverManager.recover(); 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 1e9aae0842826..b658b37dbf349 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 @@ -36,6 +36,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.LoadTsFile; 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.dataregion.utils.TsFileTableSchemaRegistrationUtils; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; @@ -59,10 +60,12 @@ import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.nio.file.Files; import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; public class LoadTsFileAnalyzerTest { @@ -485,6 +488,52 @@ private Method getAutoCreateDatabaseMethod() throws NoSuchMethodException { "s1", TypeFactory.getType(fieldType), false, TsTableColumnCategory.FIELD))); } + @Test + public void testDetectTableModelTsFile() throws Exception { + final File dir = Files.createTempDirectory("detect-table-model").toFile(); + final File markedFile = new File(dir, "marked.tsfile"); + final File treeFile = new File(dir, "tree.tsfile"); + final File legacyTableFile = new File(dir, "legacy-table.tsfile"); + try { + // A table-model file without any TableSchema entry is still recognized by its file property + try (final TsFileIOWriter writer = new TsFileIOWriter(markedFile)) { + TsFileTableSchemaRegistrationUtils.markTableModel(writer); + writer.endFile(); + } + try (final TsFileSequenceReader reader = new TsFileSequenceReader(markedFile.getPath())) { + final Map tableSchemaMap = reader.getTableSchemaMap(); + Assert.assertTrue(tableSchemaMap.isEmpty()); + Assert.assertTrue(LoadTsFileAnalyzer.detectTableModelTsFile(reader, tableSchemaMap)); + } + + // A tree-model file has neither the property nor a TableSchema map + try (final TsFileIOWriter writer = new TsFileIOWriter(treeFile)) { + writer.endFile(); + } + try (final TsFileSequenceReader reader = new TsFileSequenceReader(treeFile.getPath())) { + final Map tableSchemaMap = reader.getTableSchemaMap(); + Assert.assertFalse(LoadTsFileAnalyzer.detectTableModelTsFile(reader, tableSchemaMap)); + } + + // A table-model file written before the property existed falls back to the TableSchema map + try (final TsFileIOWriter writer = new TsFileIOWriter(legacyTableFile)) { + writer.getSchema().registerTableSchema(new TableSchema("t1")); + writer.endFile(); + } + try (final TsFileSequenceReader reader = + new TsFileSequenceReader(legacyTableFile.getPath())) { + final Map tableSchemaMap = reader.getTableSchemaMap(); + Assert.assertFalse(tableSchemaMap.isEmpty()); + Assert.assertTrue(LoadTsFileAnalyzer.detectTableModelTsFile(reader, tableSchemaMap)); + } + } finally { + Files.deleteIfExists(markedFile.toPath()); + Files.deleteIfExists(treeFile.toPath()); + Files.deleteIfExists(legacyTableFile.toPath()); + Files.deleteIfExists(dir.toPath()); + } + } + private boolean containsDevice(final Set devices, final String... expectedSegments) { return devices.stream() .anyMatch(device -> Arrays.equals(device.getSegments(), expectedSegments)); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtilsTest.java new file mode 100644 index 0000000000000..600ddd3d8a40c --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtilsTest.java @@ -0,0 +1,204 @@ +/* + * 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.dataregion.utils; + +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TimeColumnSchema; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.StringArrayDeviceID; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.file.metadata.TsFileMetadata; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.write.schema.Schema; +import org.apache.tsfile.write.writer.TsFileIOWriter; +import org.junit.After; +import org.junit.Test; +import org.mockito.AdditionalMatchers; +import org.mockito.Mockito; + +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Collections; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +public class TsFileTableSchemaRegistrationUtilsTest { + + private static final String DATABASE = "schema_registration_test"; + + @After + public void tearDown() { + DataNodeTableCache.getInstance().invalid(DATABASE); + } + + @Test + public void testRegistersMissingSchemaWithoutAddingTimeColumn() { + final String tableName = "t1"; + final TsTable table = new TsTable(tableName); + table.addColumnSchema(new TimeColumnSchema("time", TSDataType.TIMESTAMP)); + table.addColumnSchema(new TagColumnSchema("tag1", TSDataType.STRING)); + registerTable(tableName, table); + + final Schema schema = new Schema(); + final TsFileIOWriter writer = mockWriter(schema, tableName); + TsFileTableSchemaRegistrationUtils.registerTableSchemaIfAbsent(writer, DATABASE, tableName); + + final TableSchema registered = schema.getTableSchemaMap().get(tableName); + assertNotNull(registered); + assertEquals(1, registered.getColumnSchemas().size()); + assertEquals(ColumnCategory.TAG, registered.getColumnTypes().get(0)); + assertFalse(registered.getColumnTypes().contains(ColumnCategory.TIME)); + } + + @Test + public void testTimeOnlyTableStillGetsEmptySchemaEntry() { + final String tableName = "time_only_table"; + final TsTable table = new TsTable(tableName); + table.addColumnSchema(new TimeColumnSchema("time", TSDataType.TIMESTAMP)); + registerTable(tableName, table); + + final Schema schema = new Schema(); + final TsFileIOWriter writer = mockWriter(schema, tableName); + TsFileTableSchemaRegistrationUtils.registerTableSchemaIfAbsent(writer, DATABASE, tableName); + + final TableSchema registered = schema.getTableSchemaMap().get(tableName); + assertNotNull(registered); + assertTrue(registered.getColumnTypes().isEmpty()); + } + + @Test + public void testDroppedTableIsNotTouched() { + final String tableName = "dropped_table"; + final Schema schema = new Schema(); + final TsFileIOWriter writer = mockWriter(schema, tableName); + + TsFileTableSchemaRegistrationUtils.registerTableSchemaIfAbsent(writer, DATABASE, tableName); + + // The storage path does not describe a table it cannot read. The file stays recognizable + // through the table-model file property instead. + assertTrue(schema.getTableSchemaMap().isEmpty()); + } + + @Test + public void testExistingSchemaIsNotRewritten() { + final String tableName = "existing_table"; + final Schema schema = new Schema(); + final TableSchema existingSchema = new TableSchema(tableName); + schema.registerTableSchema(existingSchema); + final TsFileIOWriter writer = mockWriter(schema, tableName); + + TsFileTableSchemaRegistrationUtils.registerTableSchemaIfAbsent(writer, DATABASE, tableName); + + assertSame(existingSchema, schema.getTableSchemaMap().get(tableName)); + Mockito.verify(writer, Mockito.never()) + .addTsFileProperty(Mockito.anyString(), Mockito.any(byte[].class)); + } + + @Test + public void testDatabaseNameIsOnlyUsedAsCacheKey() { + // The model of the region is known by the caller (the owning DataRegion), so this method must + // not guess it from the name: a database whose name looks like a tree-model one, but that does + // own the table, has to be described as well. + final String database = "root.tree_named_db"; + final String tableName = "table_on_root_named_db"; + final TsTable table = new TsTable(tableName); + table.addColumnSchema(new TimeColumnSchema("time", TSDataType.TIMESTAMP)); + table.addColumnSchema(new TagColumnSchema("tag1", TSDataType.STRING)); + DataNodeTableCache.getInstance().preUpdateTable(database, table, null); + DataNodeTableCache.getInstance().commitUpdateTable(database, tableName, null, null); + try { + final Schema schema = new Schema(); + + TsFileTableSchemaRegistrationUtils.registerTableSchemaIfAbsent( + mockWriter(schema, tableName), database, tableName); + + assertNotNull(schema.getTableSchemaMap().get(tableName)); + } finally { + DataNodeTableCache.getInstance().invalid(database); + } + } + + @Test + public void testMarkTableModelWritesSingleProperty() { + final TsFileIOWriter writer = mockWriter(new Schema(), "t1"); + + TsFileTableSchemaRegistrationUtils.markTableModel(writer); + + Mockito.verify(writer) + .addTsFileProperty( + Mockito.eq(TsFileTableSchemaRegistrationUtils.TABLE_MODEL_PROPERTY), + AdditionalMatchers.aryEq("true".getBytes(StandardCharsets.UTF_8))); + } + + @Test + public void testTableModelPropertyIsPersistedWithoutTableSchema() throws IOException { + final Path tempDir = Files.createTempDirectory("table-model-marker"); + final File tsFile = tempDir.resolve("marker.tsfile").toFile(); + try { + try (final TsFileIOWriter writer = new TsFileIOWriter(tsFile)) { + TsFileTableSchemaRegistrationUtils.markTableModel(writer); + writer.endFile(); + } + + try (final TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) { + final TsFileMetadata metadata = reader.readFileMetadata(); + // The single property is enough to recognize the file as a table-model TsFile, even when + // the TableSchema map is empty. + assertTrue(TsFileTableSchemaRegistrationUtils.isTableModel(metadata)); + assertArrayEquals( + "true".getBytes(StandardCharsets.UTF_8), + metadata + .getTsFileProperties() + .get(TsFileTableSchemaRegistrationUtils.TABLE_MODEL_PROPERTY)); + assertTrue(reader.getTableSchemaMap().isEmpty()); + } + } finally { + Files.deleteIfExists(tsFile.toPath()); + Files.deleteIfExists(tempDir); + } + } + + private static void registerTable(final String tableName, final TsTable table) { + DataNodeTableCache.getInstance().preUpdateTable(DATABASE, table, null); + DataNodeTableCache.getInstance().commitUpdateTable(DATABASE, tableName, null, null); + } + + private static TsFileIOWriter mockWriter(final Schema schema, final String tableName) { + final TsFileIOWriter writer = Mockito.mock(TsFileIOWriter.class); + Mockito.when(writer.getSchema()).thenReturn(schema); + final IDeviceID device = new StringArrayDeviceID(tableName, "tag"); + Mockito.when(writer.getDeviceChunkMetadataMap()) + .thenReturn(Collections.singletonMap(device, Collections.emptyList())); + return writer; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java index cecf04e6781d7..2b268dbc8d16f 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManagerTest.java @@ -407,7 +407,7 @@ private void recoverFromOldWalAndCheck(boolean withBrokenTsFile) throws Exceptio tsFileWithWALResource = new TsFileResource(fileWithWAL); UnsealedTsFileRecoverPerformer recoverPerformer = new UnsealedTsFileRecoverPerformer( - tsFileWithWALResource, true, performer -> assertFalse(performer.canWrite())); + tsFileWithWALResource, true, false, performer -> assertFalse(performer.canWrite())); recoverManager.addRecoverPerformer(recoverPerformer); recoverListeners.add(recoverPerformer.getRecoverListener()); // recover @@ -563,7 +563,7 @@ private List prepareCrashedTsFile() tsFileWithWALResource = new TsFileResource(fileWithWAL); UnsealedTsFileRecoverPerformer recoverPerformer = new UnsealedTsFileRecoverPerformer( - tsFileWithWALResource, true, performer -> assertFalse(performer.canWrite())); + tsFileWithWALResource, true, false, performer -> assertFalse(performer.canWrite())); recoverManager.addRecoverPerformer(recoverPerformer); recoverListeners.add(recoverPerformer.getRecoverListener()); @@ -573,7 +573,7 @@ private List prepareCrashedTsFile() tsFileWithoutWALResource = new TsFileResource(fileWithoutWAL); recoverPerformer = new UnsealedTsFileRecoverPerformer( - tsFileWithoutWALResource, true, performer -> assertFalse(performer.canWrite())); + tsFileWithoutWALResource, true, false, performer -> assertFalse(performer.canWrite())); recoverManager.addRecoverPerformer(recoverPerformer); recoverListeners.add(recoverPerformer.getRecoverListener()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/EncryptedUnsealedTsFileRecoverPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/EncryptedUnsealedTsFileRecoverPerformerTest.java index 8ba805f1bb995..b13c7c9d4b45f 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/EncryptedUnsealedTsFileRecoverPerformerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/EncryptedUnsealedTsFileRecoverPerformerTest.java @@ -164,7 +164,7 @@ private void assertNoDataRecovery(boolean baseHeaderOnly, boolean aligned, boole resource = new TsFileResource(file); byte[] recoveredFileId; try (UnsealedTsFileRecoverPerformer performer = - new UnsealedTsFileRecoverPerformer(resource, true, ignored -> {})) { + new UnsealedTsFileRecoverPerformer(resource, true, false, ignored -> {})) { performer.startRecovery(); assertTrue(performer.hasCrashed()); EncryptParameter parameter = performer.getWriter().getEncryptParameter(); @@ -191,7 +191,7 @@ public void testProviderFailureDoesNotModifyUninitializedFiles() throws Exceptio Files.write(file.toPath(), original); resource = new TsFileResource(file); try (UnsealedTsFileRecoverPerformer performer = - new UnsealedTsFileRecoverPerformer(resource, true, ignored -> {})) { + new UnsealedTsFileRecoverPerformer(resource, true, false, ignored -> {})) { assertThrows(RuntimeException.class, performer::startRecovery); assertNull(performer.getWriter()); } @@ -212,7 +212,7 @@ public void testPartialHeadersCannotBeReinitialized() throws Exception { Files.write(file.toPath(), original); resource = new TsFileResource(file); try (UnsealedTsFileRecoverPerformer performer = - new UnsealedTsFileRecoverPerformer(resource, true, ignored -> {})) { + new UnsealedTsFileRecoverPerformer(resource, true, false, ignored -> {})) { assertThrows(Exception.class, performer::startRecovery); } assertArrayEquals(original, Files.readAllBytes(file.toPath())); @@ -236,7 +236,7 @@ public void testValidEncryptionHeaderWithoutDataKeepsOriginalContext() throws Ex } resource = new TsFileResource(file); try (UnsealedTsFileRecoverPerformer performer = - new UnsealedTsFileRecoverPerformer(resource, true, ignored -> {})) { + new UnsealedTsFileRecoverPerformer(resource, true, false, ignored -> {})) { performer.startRecovery(); assertArrayEquals(oldFileId, performer.getWriter().getEncryptParameter().getFileCryptoId()); performer.redoLog(firstEntry(false, false)); @@ -368,7 +368,7 @@ private void assertRecovery(boolean aligned, boolean snapshot) throws Exception resource = new TsFileResource(file); try (UnsealedTsFileRecoverPerformer performer = - new UnsealedTsFileRecoverPerformer(resource, true, ignored -> {})) { + new UnsealedTsFileRecoverPerformer(resource, true, false, ignored -> {})) { performer.startRecovery(); assertTrue(performer.hasCrashed()); assertArrayEquals( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformerTest.java index 8d43c77970cd4..3f6f8f7d1c711 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/UnsealedTsFileRecoverPerformerTest.java @@ -145,7 +145,7 @@ public void testRedoInsertPlan() throws Exception { // vsg processor is used to test IdTable, don't test IdTable here try (UnsealedTsFileRecoverPerformer recoverPerformer = new UnsealedTsFileRecoverPerformer( - tsFileResource, true, performer -> assertFalse(performer.canWrite()))) { + tsFileResource, true, false, performer -> assertFalse(performer.canWrite()))) { recoverPerformer.startRecovery(); assertTrue(recoverPerformer.hasCrashed()); assertTrue(recoverPerformer.canWrite()); @@ -201,9 +201,11 @@ public void testRedoRelationalInsertPlan() throws Exception { WALEntry walEntry = new WALInfoEntry(fakeMemTableId, insertTabletNode); // recover tsFileResource = new TsFileResource(file); + // the owner of this TsFile is a table-model DataRegion, so pass its model explicitly instead of + // letting the performer derive the model from the database name try (UnsealedTsFileRecoverPerformer recoverPerformer = new UnsealedTsFileRecoverPerformer( - tsFileResource, true, performer -> assertFalse(performer.canWrite()))) { + tsFileResource, true, true, performer -> assertFalse(performer.canWrite()))) { recoverPerformer.startRecovery(); assertTrue(recoverPerformer.hasCrashed()); assertTrue(recoverPerformer.canWrite()); @@ -240,7 +242,7 @@ public void testRedoDeletePlan() throws Exception { // vsg processor is used to test IdTable, don't test IdTable here try (UnsealedTsFileRecoverPerformer recoverPerformer = new UnsealedTsFileRecoverPerformer( - tsFileResource, true, performer -> assertFalse(performer.canWrite()))) { + tsFileResource, true, false, performer -> assertFalse(performer.canWrite()))) { recoverPerformer.startRecovery(); assertTrue(recoverPerformer.hasCrashed()); assertTrue(recoverPerformer.canWrite()); @@ -367,7 +369,7 @@ public void testRecoverNullInsertRowPlan() throws Exception { // vsg processor is used to test IdTable, don't test IdTable here try (UnsealedTsFileRecoverPerformer recoverPerformer = new UnsealedTsFileRecoverPerformer( - tsFileResource, true, performer -> assertFalse(performer.canWrite()))) { + tsFileResource, true, false, performer -> assertFalse(performer.canWrite()))) { recoverPerformer.startRecovery(); assertTrue(recoverPerformer.hasCrashed()); assertTrue(recoverPerformer.canWrite()); @@ -431,7 +433,8 @@ public void testRecoverDuplicate() WALEntry normalWalEntry = new WALInfoEntry(fakeMemTableId++, insertRowNode2); try (UnsealedTsFileRecoverPerformer performer = - new UnsealedTsFileRecoverPerformer(tsFileResource, true, p -> assertFalse(p.canWrite()))) { + new UnsealedTsFileRecoverPerformer( + tsFileResource, true, false, p -> assertFalse(p.canWrite()))) { performer.startRecovery(); performer.redoLog(duplicateMemTableSnapshotWalEntry); performer.redoLog(duplicateWalEntry);