From 2c6e96fa9c27d29e62fd0e115de0a897f27ac690 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Tue, 3 Nov 2026 11:46:56 +0800 Subject: [PATCH 1/7] Fix table schema registration for table-model TsFiles Keep the TableSchema map entry for every table that has chunk data in a TsFile, even when the schema has no FIELD or TIME column. Restore missing entries during flush and recovery, and retain them during compaction so table-model files do not degrade into tree-model files. Add unit tests and a two-cluster integration test covering compaction and LOAD. --- ...oTDBLoadStorageTsFileWithCompactionIT.java | 265 ++++++++++++++++++ .../execute/utils/CompactionTableSchema.java | 68 ++++- .../compaction/io/CompactionTsFileWriter.java | 12 +- .../dataregion/memtable/TsFileProcessor.java | 6 + .../TsFileTableSchemaRegistrationUtils.java | 84 ++++++ .../file/UnsealedTsFileRecoverPerformer.java | 22 +- .../CompactionTableSchemaCollectorTest.java | 52 ++++ .../utils/CompactionTestFileWriterTest.java | 20 ++ ...sFileTableSchemaRegistrationUtilsTest.java | 102 +++++++ 9 files changed, 606 insertions(+), 25 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadStorageTsFileWithCompactionIT.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtils.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtilsTest.java 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..34429eab20d5a --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadStorageTsFileWithCompactionIT.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.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 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)); + } + } + + 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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionTableSchema.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionTableSchema.java index 9a79a2db83930..6fd11874914e8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionTableSchema.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionTableSchema.java @@ -46,31 +46,60 @@ public boolean merge(TableSchema tableSchema) { tableName, tableSchema.getTableName())); } - // filter id columns - List otherSchemaColumnSchemas = tableSchema.getColumnSchemas(); - List otherSchemaColumnCategories = tableSchema.getColumnTypes(); - List idColumnSchemasToMerge = new ArrayList<>(); + + final List otherSchemaColumnSchemas = tableSchema.getColumnSchemas(); + final List otherSchemaColumnCategories = tableSchema.getColumnTypes(); + final List idColumnSchemasToMerge = new ArrayList<>(); + IMeasurementSchema timeColumnToMerge = null; for (int i = 0; i < otherSchemaColumnCategories.size(); i++) { - ColumnCategory columnCategory = otherSchemaColumnCategories.get(i); - if (columnCategory != ColumnCategory.TAG) { - continue; + final ColumnCategory columnCategory = otherSchemaColumnCategories.get(i); + if (columnCategory == ColumnCategory.TIME) { + // A TableSchema must contain at most one TIME column. Keep it so that a + // time-only table remains recognizable as a table-model TsFile. + if (timeColumnToMerge != null) { + return false; + } + timeColumnToMerge = otherSchemaColumnSchemas.get(i); + } else if (columnCategory == ColumnCategory.TAG) { + idColumnSchemasToMerge.add(otherSchemaColumnSchemas.get(i)); + } + } + + if (timeColumnToMerge != null) { + final int existingTimeColumnIndex = findColumnIndex(ColumnCategory.TIME); + if (existingTimeColumnIndex < 0) { + // Keep TIME before the id columns, matching the schema order produced by + // TsFileTableSchemaUtil. + measurementSchemas.add(0, timeColumnToMerge); + columnCategories.add(0, ColumnCategory.TIME); + } else if (!isSameTimeColumn( + measurementSchemas.get(existingTimeColumnIndex), timeColumnToMerge)) { + return false; + } + } + + // Only TAG columns are id columns. Filter them out from the target schema so TIME + // (and any FIELD already appended by update()) does not participate in the prefix check. + final List currentIdColumns = new ArrayList<>(); + for (int i = 0; i < columnCategories.size(); i++) { + if (columnCategories.get(i) == ColumnCategory.TAG) { + currentIdColumns.add(measurementSchemas.get(i)); } - idColumnSchemasToMerge.add(otherSchemaColumnSchemas.get(i)); } // check id column prefix - int prefixLength = Math.min(this.columnCategories.size(), idColumnSchemasToMerge.size()); + final int prefixLength = Math.min(currentIdColumns.size(), idColumnSchemasToMerge.size()); for (int i = 0; i < prefixLength; i++) { - IMeasurementSchema idColumnToMerge = idColumnSchemasToMerge.get(i); - IMeasurementSchema currentIdColumn = measurementSchemas.get(i); + final IMeasurementSchema idColumnToMerge = idColumnSchemasToMerge.get(i); + final IMeasurementSchema currentIdColumn = currentIdColumns.get(i); if (!idColumnToMerge.getMeasurementName().equals(currentIdColumn.getMeasurementName())) { return false; } } // add new id columns - List newIdColumns = + final List newIdColumns = idColumnSchemasToMerge.subList(prefixLength, idColumnSchemasToMerge.size()); for (IMeasurementSchema newIdColumn : newIdColumns) { columnCategories.add(ColumnCategory.TAG); @@ -79,6 +108,21 @@ public boolean merge(TableSchema tableSchema) { return true; } + private int findColumnIndex(final ColumnCategory columnCategory) { + for (int i = 0; i < columnCategories.size(); i++) { + if (columnCategories.get(i) == columnCategory) { + return i; + } + } + return -1; + } + + private boolean isSameTimeColumn( + final IMeasurementSchema current, final IMeasurementSchema incoming) { + return current.getMeasurementName().equalsIgnoreCase(incoming.getMeasurementName()) + && current.getType() == incoming.getType(); + } + public CompactionTableSchema copy() { CompactionTableSchema tableSchema = new CompactionTableSchema(this.tableName); tableSchema.measurementSchemas = new ArrayList<>(this.measurementSchemas); 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..6f6f36cf41d1d 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 @@ -44,10 +44,12 @@ import java.io.File; import java.io.IOException; import java.io.Serializable; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; public class CompactionTsFileWriter extends TsFileIOWriter { CompactionType type; @@ -55,6 +57,7 @@ public class CompactionTsFileWriter extends TsFileIOWriter { private volatile boolean isWritingAligned = false; private boolean isEmptyTargetFile = true; private IDeviceID currentDeviceId; + private final Set tablesWithChunkGroups = new HashSet<>(); private EncryptParameter firstEncryptParameter; @@ -167,6 +170,9 @@ public void endChunkGroup() throws IOException { TableSchema tableSchema = getSchema().getTableSchemaMap().get(tableName); boolean generateTableSchemaForCurrentChunkGroup = tableSchema != null; setGenerateTableSchema(generateTableSchemaForCurrentChunkGroup); + if (generateTableSchemaForCurrentChunkGroup) { + tablesWithChunkGroups.add(tableName); + } super.endChunkGroup(); currentDeviceId = null; } @@ -199,7 +205,11 @@ private void removeUnusedTableSchema() { while (iterator.hasNext()) { final Map.Entry entry = iterator.next(); final List columnTypes = entry.getValue().getColumnTypes(); - if (hasFieldOrTimeColumn(columnTypes)) { + // Keep the TableSchema of a table that wrote data in this target file even when the schema + // has no FIELD/TIME column. Do not synthesize columns here: the table-name entry alone is + // the table-model marker LOAD relies on, and removing it makes a table TsFile look like a + // tree TsFile. + if (hasFieldOrTimeColumn(columnTypes) || tablesWithChunkGroups.contains(entry.getKey())) { continue; } iterator.remove(); 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..d70bba5d1ff66 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; @@ -2233,6 +2234,11 @@ private void endFile() throws IOException, TsFileProcessorException { if (logger.isDebugEnabled()) { logger.debug(StorageEngineMessages.START_TO_END_FILE, tsFileResource); } + // A table-model TsFile is only recognizable by its non-empty TableSchema map. Chunks may have + // been flushed before registerToTsFile() was called for their table, so repair the map from the + // chunks and the table cache here rather than silently writing a tree-model-looking file. + TsFileTableSchemaRegistrationUtils.ensureTableSchemasForExistingChunks( + writer, dataRegionInfo.getDataRegion().getDatabaseName()); 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..cda0abe4e6006 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtils.java @@ -0,0 +1,84 @@ +/* + * 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.commons.utils.PathUtils; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; + +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; + +/** + * Ensures that a table-model TsFile retains a TableSchema for every table that has data in it. + * + *

For a table-model TsFile, a non-empty table schema map is the marker that distinguishes it + * from a tree-model TsFile. This class deliberately only changes map membership: it never adds, + * removes, or rewrites columns in an existing TableSchema. In particular, an empty schema (for + * example, one produced from a time-only table by the existing schema converter) is still written + * as-is so that the table remains identifiable. + */ +public final class TsFileTableSchemaRegistrationUtils { + + private TsFileTableSchemaRegistrationUtils() {} + + /** + * Registers the missing TableSchema for every table that already has chunks in the writer. + * + *

This is deliberately called both when a normal flush ends a file and when an unsealed file + * is recovered: in both cases chunks may already be present while the schema map is empty or + * incomplete (for example because a table was written before the table schema was registered, or + * because the WAL entry for an already-flushed chunk has been removed). + */ + public static void ensureTableSchemasForExistingChunks( + final TsFileIOWriter writer, final String databaseName) { + if (writer == null || !PathUtils.isTableModelDatabase(databaseName)) { + return; + } + for (final IDeviceID deviceId : writer.getDeviceChunkMetadataMap().keySet()) { + registerTableSchemaIfAbsent(writer, databaseName, deviceId.getTableName()); + } + } + + /** + * Registers the TableSchema for {@code tableName} if it is absent. A table that has already been + * dropped still gets an empty schema entry so that the chunks already present in the file do not + * silently turn it into a tree-model TsFile. + */ + public static void registerTableSchemaIfAbsent( + final TsFileIOWriter writer, final String databaseName, final String tableName) { + if (writer == null || tableName == null) { + return; + } + if (writer.getSchema().getTableSchemaMap().containsKey(tableName)) { + return; + } + + final TsTable table = DataNodeTableCache.getInstance().getTable(databaseName, tableName, false); + writer + .getSchema() + .registerTableSchema( + table == null + ? new TableSchema(tableName) + : 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..63c86a7ba8544 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,7 +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; @@ -29,7 +28,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 +41,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; @@ -113,6 +112,12 @@ public void startRecovery() throws DataRegionException, IOException { super.recoverWithWriter(); if (hasCrashed()) { + // Chunks that were already flushed before the crash may no longer have a WAL entry, so + // redoLog cannot be relied on to restore their TableSchema. Register every table that + // already has chunks in this file before the file is ended again. Otherwise endFile() + // would serialize an empty table schema map and turn a table-model TsFile into a + // tree-model one. + TsFileTableSchemaRegistrationUtils.ensureTableSchemasForExistingChunks(writer, databaseName); walRedoer .getRecoveryMemTable() .setEncryptParameter(EncryptUtils.getEncryptParameter(writer.getEncryptParameter())); @@ -253,16 +258,7 @@ 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))); - } + TsFileTableSchemaRegistrationUtils.registerTableSchemaIfAbsent(writer, databaseName, tableName); } /** @@ -326,6 +322,8 @@ 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 + TsFileTableSchemaRegistrationUtils.ensureTableSchemasForExistingChunks( + writer, databaseName); writer.endFile(); tsFileResource.serialize(); FileTimeIndexCacheRecorder.getInstance().logFileTimeIndex(tsFileResource); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableSchemaCollectorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableSchemaCollectorTest.java index 97006f6d43a0d..65388015d3186 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableSchemaCollectorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableSchemaCollectorTest.java @@ -30,6 +30,8 @@ import org.junit.Test; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; public class CompactionTableSchemaCollectorTest { @@ -105,4 +107,54 @@ public void test1() { TableSchema tableSchema4 = new TableSchema("t1", measurementSchemaList4, columnTypeList4); Assert.assertFalse(compactionTableSchema.merge(tableSchema4)); } + + @Test + public void testMergeTimeColumn() { + final CompactionTableSchema compactionTableSchema = new CompactionTableSchema("t1"); + final TableSchema tableSchema = + new TableSchema( + "t1", + Arrays.asList( + new MeasurementSchema("time", TSDataType.INT64), + new MeasurementSchema("id1", TSDataType.STRING)), + Arrays.asList(ColumnCategory.TIME, ColumnCategory.TAG)); + + Assert.assertTrue(compactionTableSchema.merge(tableSchema)); + Assert.assertEquals(2, compactionTableSchema.getColumnSchemas().size()); + Assert.assertEquals(ColumnCategory.TIME, compactionTableSchema.getColumnTypes().get(0)); + Assert.assertEquals( + "time", compactionTableSchema.getColumnSchemas().get(0).getMeasurementName()); + Assert.assertEquals(ColumnCategory.TAG, compactionTableSchema.getColumnTypes().get(1)); + Assert.assertEquals( + "id1", compactionTableSchema.getColumnSchemas().get(1).getMeasurementName()); + + // Merging another schema with the same TIME column must not duplicate it. + final TableSchema tableSchema2 = + new TableSchema( + "t1", + Arrays.asList( + new MeasurementSchema("time", TSDataType.INT64), + new MeasurementSchema("id1", TSDataType.STRING), + new MeasurementSchema("id2", TSDataType.STRING)), + Arrays.asList(ColumnCategory.TIME, ColumnCategory.TAG, ColumnCategory.TAG)); + Assert.assertTrue(compactionTableSchema.merge(tableSchema2)); + Assert.assertEquals(3, compactionTableSchema.getColumnSchemas().size()); + Assert.assertEquals(ColumnCategory.TIME, compactionTableSchema.getColumnTypes().get(0)); + } + + @Test + public void testMergeTimeOnlyColumn() { + final CompactionTableSchema compactionTableSchema = new CompactionTableSchema("t2"); + final TableSchema tableSchema = + new TableSchema( + "t2", + Collections.singletonList(new MeasurementSchema("time", TSDataType.INT64)), + Collections.singletonList(ColumnCategory.TIME)); + + Assert.assertTrue(compactionTableSchema.merge(tableSchema)); + Assert.assertEquals(1, compactionTableSchema.getColumnSchemas().size()); + Assert.assertEquals(ColumnCategory.TIME, compactionTableSchema.getColumnTypes().get(0)); + Assert.assertEquals( + "time", compactionTableSchema.getColumnSchemas().get(0).getMeasurementName()); + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriterTest.java index 05f345b7bbc3a..0857fb5766f20 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriterTest.java @@ -201,6 +201,24 @@ public void testGenerateAlignedSeriesWithPoints() throws IOException { } } + @Test + public void testEmptyTableSchemaWithTimeDataIsRetained() throws IOException { + final TsFileResource resource = createEmptyFileAndResource(true); + try (final CompactionTableModelTestFileWriter writer = + new CompactionTableModelTestFileWriter(resource)) { + writer.registerTableSchema("table1", Collections.emptyList()); + writer.startChunkGroup("table1", Collections.emptyList()); + writer.generateTimeOnlyDataToCurrentDevice(new TimeRange[] {new TimeRange(1, 3)}); + writer.endChunkGroup(); + writer.endFile(); + } + try (final TsFileSequenceReader reader = + new TsFileSequenceReader(resource.getTsFile().getAbsolutePath())) { + Assert.assertTrue(reader.getTableSchemaMap().containsKey("table1")); + Assert.assertTrue(reader.getTableSchemaMap().get("table1").getColumnTypes().isEmpty()); + } + } + @Test public void testFieldlessTableSchemaWithTimeDataIsRetained() throws IOException { final TsFileResource resource = createEmptyFileAndResource(true); @@ -217,6 +235,8 @@ public void testFieldlessTableSchemaWithTimeDataIsRetained() throws IOException Assert.assertTrue(reader.getTableSchemaMap().containsKey("table1")); Assert.assertFalse( reader.getTableSchemaMap().get("table1").getColumnTypes().contains(ColumnCategory.FIELD)); + Assert.assertTrue( + reader.getTableSchemaMap().get("table1").getColumnTypes().contains(ColumnCategory.TIME)); } } } 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..4991a87602058 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtilsTest.java @@ -0,0 +1,102 @@ +/* + * 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.write.schema.Schema; +import org.apache.tsfile.write.writer.TsFileIOWriter; +import org.junit.After; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.Collections; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +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(); + TsFileTableSchemaRegistrationUtils.ensureTableSchemasForExistingChunks( + mockWriter(schema, tableName), DATABASE); + + 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(); + TsFileTableSchemaRegistrationUtils.ensureTableSchemasForExistingChunks( + mockWriter(schema, tableName), DATABASE); + + final TableSchema registered = schema.getTableSchemaMap().get(tableName); + assertNotNull(registered); + assertTrue(registered.getColumnTypes().isEmpty()); + } + + 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; + } +} From 1770a21dc4d80cc11069099d9fbf0bcda7aef684 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 4 Nov 2026 16:05:18 +0800 Subject: [PATCH 2/7] Mark table-model TsFiles with a file property and complete schema in LOAD Write a single isTableModel file property when a table-model TsFile is closed by flush, compaction, recovery and the pipe sink, so a file whose TableSchema map is empty is no longer mistaken for a tree-model file. The property is constant time and independent of the number of tables. LOAD now completes a schema entry that has no column other than time, and an entry that is missing, from the table in IoTDB, and only fails with a dedicated message when the table exists neither in the file nor in IoTDB. The storage path keeps its previous behavior: it neither synthesizes entries for dropped tables nor scans chunks during recovery. Add unit tests for the file property, the model detection and the schema completion, and an integration test covering time-only entries, missing entries and table creation. --- .../it/IoTDBLoadTableSchemaCompletionIT.java | 265 ++++++++++++++++++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 6 + .../iotdb/db/i18n/DataNodeQueryMessages.java | 6 + .../PipeTableModelTsFileBuilderV2.java | 2 + .../plan/analyze/load/LoadTsFileAnalyzer.java | 18 +- .../load/LoadTsFileTableSchemaCache.java | 60 +++- .../compaction/io/CompactionTsFileWriter.java | 5 + .../dataregion/memtable/TsFileProcessor.java | 21 +- .../TsFileTableSchemaRegistrationUtils.java | 81 ++++-- .../file/UnsealedTsFileRecoverPerformer.java | 11 +- .../analyze/load/LoadTsFileAnalyzerTest.java | 49 ++++ ...sFileTableSchemaRegistrationUtilsTest.java | 100 ++++++- 12 files changed, 572 insertions(+), 52 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTableSchemaCompletionIT.java 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..6f46fc83b5f28 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,7 @@ private void writeTabletsIntoOneFile( new MemTableFlushTask(memTable, writer, null, null, true); memTableFlushTask.syncFlushMemTable(); + 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/compaction/io/CompactionTsFileWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/io/CompactionTsFileWriter.java index 6f6f36cf41d1d..515c19ae50442 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; @@ -180,6 +181,10 @@ public void endChunkGroup() throws IOException { @Override public void endFile() throws IOException { removeUnusedTableSchema(); + if (!getSchema().getTableSchemaMap().isEmpty()) { + // A single table-model marker; readers no longer have to infer the model from the schema map. + TsFileTableSchemaRegistrationUtils.markTableModel(this); + } long beforeSize = this.getPos(); super.endFile(); long writtenDataSize = this.getPos() - beforeSize; 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 d70bba5d1ff66..d85590eb8d69f 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 @@ -160,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; @@ -247,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() @@ -270,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 = @@ -2234,11 +2244,12 @@ private void endFile() throws IOException, TsFileProcessorException { if (logger.isDebugEnabled()) { logger.debug(StorageEngineMessages.START_TO_END_FILE, tsFileResource); } - // A table-model TsFile is only recognizable by its non-empty TableSchema map. Chunks may have - // been flushed before registerToTsFile() was called for their table, so repair the map from the - // chunks and the table cache here rather than silently writing a tree-model-looking file. - TsFileTableSchemaRegistrationUtils.ensureTableSchemasForExistingChunks( - writer, dataRegionInfo.getDataRegion().getDatabaseName()); + // Mark the file as a table-model TsFile. This is a single constant-time file property: the + // TableSchema entries themselves are registered on the write path by registerToTsFile(), and + // are repaired during recovery and compaction, so no per-device scan is needed at close. + 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 index cda0abe4e6006..37e48ef2800da 100644 --- 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 @@ -24,61 +24,80 @@ import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; -import org.apache.tsfile.file.metadata.IDeviceID; -import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.file.metadata.TsFileMetadata; import org.apache.tsfile.write.writer.TsFileIOWriter; +import java.nio.charset.StandardCharsets; +import java.util.Objects; + /** - * Ensures that a table-model TsFile retains a TableSchema for every table that has data in it. + * 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. * - *

For a table-model TsFile, a non-empty table schema map is the marker that distinguishes it - * from a tree-model TsFile. This class deliberately only changes map membership: it never adds, - * removes, or rewrites columns in an existing TableSchema. In particular, an empty schema (for - * example, one produced from a time-only table by the existing schema converter) is still written - * as-is so that the table remains identifiable. + *

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() {} /** - * Registers the missing TableSchema for every table that already has chunks in the writer. - * - *

This is deliberately called both when a normal flush ends a file and when an unsealed file - * is recovered: in both cases chunks may already be present while the schema map is empty or - * incomplete (for example because a table was written before the table schema was registered, or - * because the WAL entry for an already-flushed chunk has been removed). + * Records that the file being written belongs to the table model. This is one file property, so + * the write path pays no per-table cost. */ - public static void ensureTableSchemasForExistingChunks( - final TsFileIOWriter writer, final String databaseName) { - if (writer == null || !PathUtils.isTableModelDatabase(databaseName)) { + public static void markTableModel(final TsFileIOWriter writer) { + if (writer == null) { return; } - for (final IDeviceID deviceId : writer.getDeviceChunkMetadataMap().keySet()) { - registerTableSchemaIfAbsent(writer, databaseName, deviceId.getTableName()); - } + writer.addTsFileProperty(TABLE_MODEL_PROPERTY, TABLE_MODEL_PROPERTY_VALUE); } /** - * Registers the TableSchema for {@code tableName} if it is absent. A table that has already been - * dropped still gets an empty schema entry so that the chunks already present in the file do not - * silently turn it into a tree-model TsFile. + * 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. */ public static void registerTableSchemaIfAbsent( final TsFileIOWriter writer, final String databaseName, final String tableName) { - if (writer == null || tableName == null) { + if (Objects.isNull(writer) + || Objects.isNull(tableName) + || !PathUtils.isTableModelDatabase(databaseName) + || writer.getSchema().getTableSchemaMap().containsKey(tableName)) { return; } - if (writer.getSchema().getTableSchemaMap().containsKey(tableName)) { + final TsTable table = DataNodeTableCache.getInstance().getTable(databaseName, tableName, false); + if (Objects.isNull(table)) { return; } - - final TsTable table = DataNodeTableCache.getInstance().getTable(databaseName, tableName, false); writer .getSchema() - .registerTableSchema( - table == null - ? new TableSchema(tableName) - : TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table)); + .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 63c86a7ba8544..a19aee82ca1b2 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 @@ -112,12 +112,6 @@ public void startRecovery() throws DataRegionException, IOException { super.recoverWithWriter(); if (hasCrashed()) { - // Chunks that were already flushed before the crash may no longer have a WAL entry, so - // redoLog cannot be relied on to restore their TableSchema. Register every table that - // already has chunks in this file before the file is ended again. Otherwise endFile() - // would serialize an empty table schema map and turn a table-model TsFile into a - // tree-model one. - TsFileTableSchemaRegistrationUtils.ensureTableSchemasForExistingChunks(writer, databaseName); walRedoer .getRecoveryMemTable() .setEncryptParameter(EncryptUtils.getEncryptParameter(writer.getEncryptParameter())); @@ -322,8 +316,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 - TsFileTableSchemaRegistrationUtils.ensureTableSchemasForExistingChunks( - writer, databaseName); + 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/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 index 4991a87602058..37157b76dd40b 100644 --- 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 @@ -29,17 +29,27 @@ 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 { @@ -60,8 +70,8 @@ public void testRegistersMissingSchemaWithoutAddingTimeColumn() { registerTable(tableName, table); final Schema schema = new Schema(); - TsFileTableSchemaRegistrationUtils.ensureTableSchemasForExistingChunks( - mockWriter(schema, tableName), DATABASE); + final TsFileIOWriter writer = mockWriter(schema, tableName); + TsFileTableSchemaRegistrationUtils.registerTableSchemaIfAbsent(writer, DATABASE, tableName); final TableSchema registered = schema.getTableSchemaMap().get(tableName); assertNotNull(registered); @@ -78,14 +88,96 @@ public void testTimeOnlyTableStillGetsEmptySchemaEntry() { registerTable(tableName, table); final Schema schema = new Schema(); - TsFileTableSchemaRegistrationUtils.ensureTableSchemasForExistingChunks( - mockWriter(schema, tableName), DATABASE); + 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 testTreeModelDatabaseIsNotTouched() { + final String tableName = "tree_table"; + final Schema schema = new Schema(); + final TsFileIOWriter writer = mockWriter(schema, tableName); + + TsFileTableSchemaRegistrationUtils.registerTableSchemaIfAbsent( + writer, "root.tree_db", tableName); + + assertTrue(schema.getTableSchemaMap().isEmpty()); + Mockito.verify(writer, Mockito.never()) + .addTsFileProperty(Mockito.anyString(), Mockito.any(byte[].class)); + } + + @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); From 288276dfd3f1ae5635f4b36046ee745061eaad36 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 4 Nov 2026 16:17:23 +0800 Subject: [PATCH 3/7] Restore the previous TableSchema merge logic in table-model compaction CompactionTableSchema.merge behaves exactly as before: it merges id columns and never keeps the TIME column. Keeping the table-name entry of a table that wrote data into the target file is enough for a compacted file to stay identifiable, so the merge decision does not need to change. Restore the corresponding unit tests. --- .../execute/utils/CompactionTableSchema.java | 68 ++++--------------- .../CompactionTableSchemaCollectorTest.java | 52 -------------- .../utils/CompactionTestFileWriterTest.java | 20 ------ 3 files changed, 12 insertions(+), 128 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionTableSchema.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionTableSchema.java index 6fd11874914e8..9a79a2db83930 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionTableSchema.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionTableSchema.java @@ -46,60 +46,31 @@ public boolean merge(TableSchema tableSchema) { tableName, tableSchema.getTableName())); } - - final List otherSchemaColumnSchemas = tableSchema.getColumnSchemas(); - final List otherSchemaColumnCategories = tableSchema.getColumnTypes(); - final List idColumnSchemasToMerge = new ArrayList<>(); - IMeasurementSchema timeColumnToMerge = null; + // filter id columns + List otherSchemaColumnSchemas = tableSchema.getColumnSchemas(); + List otherSchemaColumnCategories = tableSchema.getColumnTypes(); + List idColumnSchemasToMerge = new ArrayList<>(); for (int i = 0; i < otherSchemaColumnCategories.size(); i++) { - final ColumnCategory columnCategory = otherSchemaColumnCategories.get(i); - if (columnCategory == ColumnCategory.TIME) { - // A TableSchema must contain at most one TIME column. Keep it so that a - // time-only table remains recognizable as a table-model TsFile. - if (timeColumnToMerge != null) { - return false; - } - timeColumnToMerge = otherSchemaColumnSchemas.get(i); - } else if (columnCategory == ColumnCategory.TAG) { - idColumnSchemasToMerge.add(otherSchemaColumnSchemas.get(i)); - } - } - - if (timeColumnToMerge != null) { - final int existingTimeColumnIndex = findColumnIndex(ColumnCategory.TIME); - if (existingTimeColumnIndex < 0) { - // Keep TIME before the id columns, matching the schema order produced by - // TsFileTableSchemaUtil. - measurementSchemas.add(0, timeColumnToMerge); - columnCategories.add(0, ColumnCategory.TIME); - } else if (!isSameTimeColumn( - measurementSchemas.get(existingTimeColumnIndex), timeColumnToMerge)) { - return false; - } - } - - // Only TAG columns are id columns. Filter them out from the target schema so TIME - // (and any FIELD already appended by update()) does not participate in the prefix check. - final List currentIdColumns = new ArrayList<>(); - for (int i = 0; i < columnCategories.size(); i++) { - if (columnCategories.get(i) == ColumnCategory.TAG) { - currentIdColumns.add(measurementSchemas.get(i)); + ColumnCategory columnCategory = otherSchemaColumnCategories.get(i); + if (columnCategory != ColumnCategory.TAG) { + continue; } + idColumnSchemasToMerge.add(otherSchemaColumnSchemas.get(i)); } // check id column prefix - final int prefixLength = Math.min(currentIdColumns.size(), idColumnSchemasToMerge.size()); + int prefixLength = Math.min(this.columnCategories.size(), idColumnSchemasToMerge.size()); for (int i = 0; i < prefixLength; i++) { - final IMeasurementSchema idColumnToMerge = idColumnSchemasToMerge.get(i); - final IMeasurementSchema currentIdColumn = currentIdColumns.get(i); + IMeasurementSchema idColumnToMerge = idColumnSchemasToMerge.get(i); + IMeasurementSchema currentIdColumn = measurementSchemas.get(i); if (!idColumnToMerge.getMeasurementName().equals(currentIdColumn.getMeasurementName())) { return false; } } // add new id columns - final List newIdColumns = + List newIdColumns = idColumnSchemasToMerge.subList(prefixLength, idColumnSchemasToMerge.size()); for (IMeasurementSchema newIdColumn : newIdColumns) { columnCategories.add(ColumnCategory.TAG); @@ -108,21 +79,6 @@ public boolean merge(TableSchema tableSchema) { return true; } - private int findColumnIndex(final ColumnCategory columnCategory) { - for (int i = 0; i < columnCategories.size(); i++) { - if (columnCategories.get(i) == columnCategory) { - return i; - } - } - return -1; - } - - private boolean isSameTimeColumn( - final IMeasurementSchema current, final IMeasurementSchema incoming) { - return current.getMeasurementName().equalsIgnoreCase(incoming.getMeasurementName()) - && current.getType() == incoming.getType(); - } - public CompactionTableSchema copy() { CompactionTableSchema tableSchema = new CompactionTableSchema(this.tableName); tableSchema.measurementSchemas = new ArrayList<>(this.measurementSchemas); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableSchemaCollectorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableSchemaCollectorTest.java index 65388015d3186..97006f6d43a0d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableSchemaCollectorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableSchemaCollectorTest.java @@ -30,8 +30,6 @@ import org.junit.Test; import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; import java.util.List; public class CompactionTableSchemaCollectorTest { @@ -107,54 +105,4 @@ public void test1() { TableSchema tableSchema4 = new TableSchema("t1", measurementSchemaList4, columnTypeList4); Assert.assertFalse(compactionTableSchema.merge(tableSchema4)); } - - @Test - public void testMergeTimeColumn() { - final CompactionTableSchema compactionTableSchema = new CompactionTableSchema("t1"); - final TableSchema tableSchema = - new TableSchema( - "t1", - Arrays.asList( - new MeasurementSchema("time", TSDataType.INT64), - new MeasurementSchema("id1", TSDataType.STRING)), - Arrays.asList(ColumnCategory.TIME, ColumnCategory.TAG)); - - Assert.assertTrue(compactionTableSchema.merge(tableSchema)); - Assert.assertEquals(2, compactionTableSchema.getColumnSchemas().size()); - Assert.assertEquals(ColumnCategory.TIME, compactionTableSchema.getColumnTypes().get(0)); - Assert.assertEquals( - "time", compactionTableSchema.getColumnSchemas().get(0).getMeasurementName()); - Assert.assertEquals(ColumnCategory.TAG, compactionTableSchema.getColumnTypes().get(1)); - Assert.assertEquals( - "id1", compactionTableSchema.getColumnSchemas().get(1).getMeasurementName()); - - // Merging another schema with the same TIME column must not duplicate it. - final TableSchema tableSchema2 = - new TableSchema( - "t1", - Arrays.asList( - new MeasurementSchema("time", TSDataType.INT64), - new MeasurementSchema("id1", TSDataType.STRING), - new MeasurementSchema("id2", TSDataType.STRING)), - Arrays.asList(ColumnCategory.TIME, ColumnCategory.TAG, ColumnCategory.TAG)); - Assert.assertTrue(compactionTableSchema.merge(tableSchema2)); - Assert.assertEquals(3, compactionTableSchema.getColumnSchemas().size()); - Assert.assertEquals(ColumnCategory.TIME, compactionTableSchema.getColumnTypes().get(0)); - } - - @Test - public void testMergeTimeOnlyColumn() { - final CompactionTableSchema compactionTableSchema = new CompactionTableSchema("t2"); - final TableSchema tableSchema = - new TableSchema( - "t2", - Collections.singletonList(new MeasurementSchema("time", TSDataType.INT64)), - Collections.singletonList(ColumnCategory.TIME)); - - Assert.assertTrue(compactionTableSchema.merge(tableSchema)); - Assert.assertEquals(1, compactionTableSchema.getColumnSchemas().size()); - Assert.assertEquals(ColumnCategory.TIME, compactionTableSchema.getColumnTypes().get(0)); - Assert.assertEquals( - "time", compactionTableSchema.getColumnSchemas().get(0).getMeasurementName()); - } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriterTest.java index 0857fb5766f20..05f345b7bbc3a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTestFileWriterTest.java @@ -201,24 +201,6 @@ public void testGenerateAlignedSeriesWithPoints() throws IOException { } } - @Test - public void testEmptyTableSchemaWithTimeDataIsRetained() throws IOException { - final TsFileResource resource = createEmptyFileAndResource(true); - try (final CompactionTableModelTestFileWriter writer = - new CompactionTableModelTestFileWriter(resource)) { - writer.registerTableSchema("table1", Collections.emptyList()); - writer.startChunkGroup("table1", Collections.emptyList()); - writer.generateTimeOnlyDataToCurrentDevice(new TimeRange[] {new TimeRange(1, 3)}); - writer.endChunkGroup(); - writer.endFile(); - } - try (final TsFileSequenceReader reader = - new TsFileSequenceReader(resource.getTsFile().getAbsolutePath())) { - Assert.assertTrue(reader.getTableSchemaMap().containsKey("table1")); - Assert.assertTrue(reader.getTableSchemaMap().get("table1").getColumnTypes().isEmpty()); - } - } - @Test public void testFieldlessTableSchemaWithTimeDataIsRetained() throws IOException { final TsFileResource resource = createEmptyFileAndResource(true); @@ -235,8 +217,6 @@ public void testFieldlessTableSchemaWithTimeDataIsRetained() throws IOException Assert.assertTrue(reader.getTableSchemaMap().containsKey("table1")); Assert.assertFalse( reader.getTableSchemaMap().get("table1").getColumnTypes().contains(ColumnCategory.FIELD)); - Assert.assertTrue( - reader.getTableSchemaMap().get("table1").getColumnTypes().contains(ColumnCategory.TIME)); } } } From 757b8186710030e1ee7873de4d57fbdd6df70d7d Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 4 Nov 2026 16:18:43 +0800 Subject: [PATCH 4/7] Stop keeping empty TableSchema entries in table-model compaction Compaction drops a TableSchema again when it has no FIELD or TIME column, as it did before this change. The target file stays identifiable through the isTableModel file property, so the writer only records that property, read before the unused entries are removed. --- .../compaction/io/CompactionTsFileWriter.java | 19 ++++++------------- 1 file changed, 6 insertions(+), 13 deletions(-) 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 515c19ae50442..b170075ff11e5 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 @@ -45,12 +45,10 @@ import java.io.File; import java.io.IOException; import java.io.Serializable; -import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Objects; -import java.util.Set; public class CompactionTsFileWriter extends TsFileIOWriter { CompactionType type; @@ -58,7 +56,6 @@ public class CompactionTsFileWriter extends TsFileIOWriter { private volatile boolean isWritingAligned = false; private boolean isEmptyTargetFile = true; private IDeviceID currentDeviceId; - private final Set tablesWithChunkGroups = new HashSet<>(); private EncryptParameter firstEncryptParameter; @@ -171,18 +168,18 @@ public void endChunkGroup() throws IOException { TableSchema tableSchema = getSchema().getTableSchemaMap().get(tableName); boolean generateTableSchemaForCurrentChunkGroup = tableSchema != null; setGenerateTableSchema(generateTableSchemaForCurrentChunkGroup); - if (generateTableSchemaForCurrentChunkGroup) { - tablesWithChunkGroups.add(tableName); - } super.endChunkGroup(); currentDeviceId = null; } @Override public void endFile() throws IOException { + // Whether the target file collects any table schema, that is, whether it belongs to the table + // model. Read before the unused entries are dropped so that a file whose entries are all + // dropped is still marked as a table-model file. + final boolean isTableModelFile = !getSchema().getTableSchemaMap().isEmpty(); removeUnusedTableSchema(); - if (!getSchema().getTableSchemaMap().isEmpty()) { - // A single table-model marker; readers no longer have to infer the model from the schema map. + if (isTableModelFile) { TsFileTableSchemaRegistrationUtils.markTableModel(this); } long beforeSize = this.getPos(); @@ -210,11 +207,7 @@ private void removeUnusedTableSchema() { while (iterator.hasNext()) { final Map.Entry entry = iterator.next(); final List columnTypes = entry.getValue().getColumnTypes(); - // Keep the TableSchema of a table that wrote data in this target file even when the schema - // has no FIELD/TIME column. Do not synthesize columns here: the table-name entry alone is - // the table-model marker LOAD relies on, and removing it makes a table TsFile look like a - // tree TsFile. - if (hasFieldOrTimeColumn(columnTypes) || tablesWithChunkGroups.contains(entry.getKey())) { + if (hasFieldOrTimeColumn(columnTypes)) { continue; } iterator.remove(); From 986edb967999367f339d65e60a15fa79d11b810c Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 4 Nov 2026 16:21:23 +0800 Subject: [PATCH 5/7] Mark every table-model TsFile written by compaction The compaction writer learns the model from the database it compacts, so the isTableModel property is written whenever the target file belongs to a table-model database, even when its TableSchema map keeps no entry. Assert the property in the compaction and LOAD integration test. --- .../IoTDBLoadStorageTsFileWithCompactionIT.java | 4 ++++ .../impl/ReadChunkCompactionPerformer.java | 4 ++++ .../compaction/io/CompactionTsFileWriter.java | 15 +++++++++------ 3 files changed, 17 insertions(+), 6 deletions(-) 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 index 34429eab20d5a..bb8c79b921b18 100644 --- 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 @@ -57,6 +57,7 @@ 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; @@ -200,6 +201,9 @@ private static void assertTableSchemaRegistered(final Path tsFile) throws IOExce 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)); } } 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 b170075ff11e5..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 @@ -56,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; @@ -92,6 +93,10 @@ public static CompactionTsFileWriter create( } } + public void setTableModel(final boolean tableModel) { + this.tableModel = tableModel; + } + public void markStartingWritingAligned() { isWritingAligned = true; } @@ -174,14 +179,12 @@ public void endChunkGroup() throws IOException { @Override public void endFile() throws IOException { - // Whether the target file collects any table schema, that is, whether it belongs to the table - // model. Read before the unused entries are dropped so that a file whose entries are all - // dropped is still marked as a table-model file. - final boolean isTableModelFile = !getSchema().getTableSchemaMap().isEmpty(); - removeUnusedTableSchema(); - if (isTableModelFile) { + // 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(); long writtenDataSize = this.getPos() - beforeSize; From f3d5421ebadf1c34cd46f31e125f3f9ee0a96d94 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Wed, 4 Nov 2026 16:27:44 +0800 Subject: [PATCH 6/7] Document that the table-model marker is written once when a file is sealed The marker is one entry of the in-memory file property map, serialized by endFile(). Spell that out at the helper and at the flush and pipe call sites so the marker is never moved to a per-record, per-tablet or per-chunk-group position. --- .../sink/util/builder/PipeTableModelTsFileBuilderV2.java | 2 ++ .../dataregion/memtable/TsFileProcessor.java | 6 +++--- .../utils/TsFileTableSchemaRegistrationUtils.java | 8 ++++++-- 3 files changed, 11 insertions(+), 5 deletions(-) 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 6f46fc83b5f28..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 @@ -340,6 +340,8 @@ 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/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index d85590eb8d69f..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 @@ -2244,9 +2244,9 @@ 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. This is a single constant-time file property: the - // TableSchema entries themselves are registered on the write path by registerToTsFile(), and - // are repaired during recovery and compaction, so no per-device scan is needed at close. + // 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); } 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 index 37e48ef2800da..402e889cd68e2 100644 --- 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 @@ -52,8 +52,12 @@ public final class TsFileTableSchemaRegistrationUtils { private TsFileTableSchemaRegistrationUtils() {} /** - * Records that the file being written belongs to the table model. This is one file property, so - * the write path pays no per-table cost. + * 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) { From 36c92013e6ebbbe100076100ed68d743cc4a6656 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 5 Nov 2026 10:27:08 +0800 Subject: [PATCH 7/7] Take the table-model flag from DataRegion during WAL recovery UnsealedTsFileRecoverPerformer derived the data model from the database name via PathUtils.isTableModelDatabase, and the table-schema registration helper repeated the same name-based check. That duplicates knowledge the owning DataRegion already holds through its isTableModel field and misclassifies a table-model region whose database name happens to start with "root.", which leaves the recovered TsFile with the chunks of a table but no TableSchema entry for them. Pass DataRegion.isTableModel into the recover performer, gate the schema registration on it, and keep databaseName only as the DataNodeTableCache lookup key. Update the recovery tests to pass the model explicitly and turn the registration-helper test into a check that the database name is never inspected to guess the model. --- .../storageengine/dataregion/DataRegion.java | 3 +- .../TsFileTableSchemaRegistrationUtils.java | 7 +++-- .../file/UnsealedTsFileRecoverPerformer.java | 10 +++++-- .../dataregion/IoTConsensusShutdownTest.java | 10 +++++-- ...sFileTableSchemaRegistrationUtilsTest.java | 28 +++++++++++++------ .../wal/recover/WALRecoverManagerTest.java | 6 ++-- ...tedUnsealedTsFileRecoverPerformerTest.java | 10 +++---- .../UnsealedTsFileRecoverPerformerTest.java | 13 +++++---- 8 files changed, 58 insertions(+), 29 deletions(-) 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/utils/TsFileTableSchemaRegistrationUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtils.java index 402e889cd68e2..78a4165efed51 100644 --- 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 @@ -21,7 +21,6 @@ import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.commons.schema.table.TsTable; -import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.tsfile.file.metadata.TsFileMetadata; @@ -87,12 +86,16 @@ public static boolean isTableModel(final TsFileMetadata metadata) { *

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) - || !PathUtils.isTableModelDatabase(databaseName) || writer.getSchema().getTableSchemaMap().containsKey(tableName)) { return; } 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 a19aee82ca1b2..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,7 +20,6 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.recover.file; import org.apache.iotdb.commons.path.MeasurementPath; -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; @@ -90,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); @@ -252,6 +252,12 @@ public void redoLog(WALEntry walEntry) { } private void registerToTsFile(String tableName) { + // 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); } 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/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/utils/TsFileTableSchemaRegistrationUtilsTest.java index 37157b76dd40b..600ddd3d8a40c 100644 --- 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 @@ -125,17 +125,27 @@ public void testExistingSchemaIsNotRewritten() { } @Test - public void testTreeModelDatabaseIsNotTouched() { - final String tableName = "tree_table"; - final Schema schema = new Schema(); - final TsFileIOWriter writer = mockWriter(schema, tableName); + 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( - writer, "root.tree_db", tableName); + TsFileTableSchemaRegistrationUtils.registerTableSchemaIfAbsent( + mockWriter(schema, tableName), database, tableName); - assertTrue(schema.getTableSchemaMap().isEmpty()); - Mockito.verify(writer, Mockito.never()) - .addTsFileProperty(Mockito.anyString(), Mockito.any(byte[].class)); + assertNotNull(schema.getTableSchemaMap().get(tableName)); + } finally { + DataNodeTableCache.getInstance().invalid(database); + } } @Test 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);