This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 26612d5fcb0c4ad338160d4dc3556c31a622183f Author: Tian Jiang <[email protected]> AuthorDate: Thu Oct 15 10:40:03 2026 +0800 tmp --- .../file/UnsealedTsFileRecoverPerformer.java | 25 +++++++++---- .../wal/recover/file/FlatDeviceRecoveryTest.java | 43 ++++++++++++++++++++-- 2 files changed, 56 insertions(+), 12 deletions(-) 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 834da656189..34429520a88 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 @@ -66,6 +66,7 @@ import java.util.concurrent.ExecutionException; import java.util.function.Consumer; import static org.apache.iotdb.commons.conf.IoTDBConstant.FILE_NAME_SEPARATOR; +import static org.apache.iotdb.commons.utils.PathUtils.isTableModelDatabase; /** * This class is used to help recover all unsealed TsFiles at zero level. There are 3 main @@ -86,6 +87,8 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform // trace result of this recovery private final WALRecoverListener recoverListener; private final String databaseName; + // The database determines the file's model, including when a table name resembles a tree path. + private final boolean isTableModel; private final String dataRegionId; public UnsealedTsFileRecoverPerformer( @@ -94,6 +97,7 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform Consumer<UnsealedTsFileRecoverPerformer> callbackAfterUnsealedTsFileRecovered) { super(tsFileResource); this.databaseName = tsFileResource.getDatabaseName(); + this.isTableModel = isTableModelDatabase(databaseName); this.dataRegionId = tsFileResource.getDataRegionId(); this.sequence = sequence; this.callbackAfterUnsealedTsFileRecovered = callbackAfterUnsealedTsFileRecovered; @@ -124,6 +128,11 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform Map<IDeviceID, List<ChunkMetadata>> deviceChunkMetaDataMap = writer.getDeviceChunkMetadataMap(); for (Map.Entry<IDeviceID, List<ChunkMetadata>> entry : deviceChunkMetaDataMap.entrySet()) { IDeviceID deviceId = entry.getKey(); + if (isTableModel) { + // A completed flush may leave no WAL to replay. Its recovered chunks still need a table + // schema and layout marker before endFile writes the replacement footer. + registerToTsFile(deviceId.getTableName()); + } List<ChunkMetadata> chunkMetadataList = entry.getValue(); // measurement -> ChunkMetadataList @@ -199,7 +208,7 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform continue; } // iterating the index, must present - if (device.isTableModel()) { + if (isTableModel) { memTable.delete( new TableDeletionEntry( new DeletionPredicate(device.getTableName(), new FullExactMatch(device)), @@ -215,8 +224,8 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform } } } - for (IDeviceID deviceID : memTable.getMemTableMap().keySet()) { - if (deviceID.isTableModel()) { + if (isTableModel) { + for (IDeviceID deviceID : memTable.getMemTableMap().keySet()) { registerToTsFile(deviceID.getTableName()); } } @@ -258,12 +267,12 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform private void registerToTsFile(String tableName) { if (tableName != null) { final TsTable table = DataNodeTableCache.getInstance().getTable(databaseName, tableName); - writer - .getSchema() - .getTableSchemaMap() - .computeIfAbsent( - tableName, t -> TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table)); if (table != null) { + writer + .getSchema() + .getTableSchemaMap() + .computeIfAbsent( + tableName, t -> TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table)); TsFileTableSchemaUtil.addFlatDeviceProperty(writer, tableName, table.isFlatDevice()); } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java index d30d6e3cda9..437890c7a44 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/FlatDeviceRecoveryTest.java @@ -81,7 +81,7 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; public class FlatDeviceRecoveryTest { - private static final String DATABASE = "root.flat_recovery"; + private static final String DATABASE = "flat_recovery"; private static final String TABLE = "flat_table"; private static final IDeviceID PHYSICAL_DEVICE = IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {TABLE}); @@ -220,6 +220,37 @@ public class FlatDeviceRecoveryTest { assertEquals(List.of("10:A:1", "100:A:1", "20:B:2"), readRecoveredFile()); } + // Recover a completed flush for a table named root.flat_table without WAL replay. Its physical + // ID resembles a tree path, but the database context must still restore its schema and flat + // marker. + @Test + public void testCrashRecoveryUsesTableDatabaseContext() throws Exception { + String tableName = "root.flat_table"; + TsTable table = new TsTable(tableName); + table.addColumnSchema(new TagColumnSchema("tag", TSDataType.STRING)); + table.addColumnSchema(new FieldColumnSchema("s", TSDataType.INT64)); + table.addProp(TsTable.FLAT_DEVICE_PROPERTY, "true"); + DataNodeTableCache.getInstance().preUpdateTable(DATABASE, table, null); + DataNodeTableCache.getInstance().commitUpdateTable(DATABASE, tableName, null); + writeUnsealedFile(tableName); + try (UnsealedTsFileRecoverPerformer performer = + new UnsealedTsFileRecoverPerformer(resource, false, ignored -> {})) { + performer.startRecovery(); + assertTrue(performer.hasCrashed()); + performer.endRecovery(); + } + resource.close(); + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getPath())) { + assertEquals(Set.of(tableName), reader.getTableSchemaMap().keySet()); + assertEquals( + List.of("tag", "s"), + reader.getTableSchemaMap().get(tableName).getColumnSchemas().stream() + .map(schema -> schema.getMeasurementName()) + .toList()); + assertTrue(TsFileTableSchemaUtil.isFlatDevice(reader.getTsFileProperties(), tableName)); + } + } + // Recover flushed A/B rows, then replay an updated A at 100 plus B at 50 and null TAG at 60. // Merge by complete TAG/time keys, retaining all five rows with the later WAL value for A. @Test @@ -300,14 +331,18 @@ public class FlatDeviceRecoveryTest { } private void writeUnsealedFile() throws Exception { + writeUnsealedFile(TABLE); + } + + private void writeUnsealedFile(String tableName) throws Exception { try (TsFileIOWriter writer = new TsFileIOWriter(file)) { writer .getSchema() .registerTableSchema( TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute( - DataNodeTableCache.getInstance().getTable(DATABASE, TABLE))); - TsFileTableSchemaUtil.addFlatDeviceProperty(writer, TABLE, true); - writer.startChunkGroup(PHYSICAL_DEVICE); + DataNodeTableCache.getInstance().getTable(DATABASE, tableName))); + TsFileTableSchemaUtil.addFlatDeviceProperty(writer, tableName, true); + writer.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {tableName})); FlatDeviceAlignedChunkWriter chunk = new FlatDeviceAlignedChunkWriter( Arrays.asList(SCHEMAS), EncryptDBUtils.getDefaultFirstEncryptParam());
