From ad71d66d71209f84d68f24c6fff04649d471520a Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 19 Oct 2026 10:00:12 +0800 Subject: [PATCH 1/2] [Test] Use consistent time units for asynchronous assertion flushes --- .../iotdb/db/it/utils/IoTDBTestUtilsIT.java | 7 ++++ .../apache/iotdb/db/it/utils/TestUtils.java | 36 +++++++++++++------ 2 files changed, 32 insertions(+), 11 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java index 80d00c3843b3d..814de53ba68e4 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/IoTDBTestUtilsIT.java @@ -37,6 +37,7 @@ import java.util.function.Consumer; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -66,6 +67,12 @@ public void testAssertAlwaysFail() { () -> TestUtils.assertAlwaysFail(EnvFactory.getEnv(), "show databases", 1)); } + @Test + public void testHalfTimeoutElapsedUsesSeconds() { + assertFalse(TestUtils.isHalfTimeoutElapsed(299_999, 600)); + assertTrue(TestUtils.isHalfTimeoutElapsed(300_001, 600)); + } + @Test public void testAssertDataAlwaysFailsOnMismatch() { final BaseEnv env = EnvFactory.getEnv(); diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java index e559fe9fb1897..79a2f1c984769 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java @@ -85,6 +85,10 @@ public class TestUtils { public static final String TIME_PRECISION_IN_NS = "ns"; + static boolean isHalfTimeoutElapsed(final long elapsedTimeInMillis, final long timeoutSeconds) { + return elapsedTimeInMillis > TimeUnit.SECONDS.toMillis(timeoutSeconds) / 2; + } + /** * Creates a statement on the write node without comparing results across DataNodes. * @@ -1702,7 +1706,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -1783,7 +1789,9 @@ public static void assertDataSizeEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -1822,7 +1830,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -1904,7 +1914,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2024,7 +2036,9 @@ public static void assertDataEventuallyOnEnv( // For IoTV2 batch mode, the pipe receiver may need to flush because the replica // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases - if (!flushed[0] && System.currentTimeMillis() - startTime > timeoutSeconds >> 1) { + if (!flushed[0] + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, timeoutSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2095,8 +2109,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime - > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, consistentSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2137,8 +2151,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime - > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, consistentSeconds)) { flushed[0] = true; statement.execute("flush"); } @@ -2192,8 +2206,8 @@ public static void assertDataAlwaysOnEnv( // sync requires tsFile to process. We flush in the middle of assertion because we // don't know when the data reaches the receiver in general cases if (!flushed[0] - && System.currentTimeMillis() - startTime - > TimeUnit.SECONDS.toMillis(consistentSeconds) / 2) { + && isHalfTimeoutElapsed( + System.currentTimeMillis() - startTime, consistentSeconds)) { flushed[0] = true; statement.execute("flush"); } From 41fd18dd6ddc2a39e7d8cd6c8145641fdfe6e5c1 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:11:45 +0800 Subject: [PATCH 2/2] [Fix] Preserve table schemas in consensus LOAD TsFiles Register table schemas before sealing consensus LOAD files, including recovered tasks, so Pipe can enumerate their table data. Cover normal loads, recovery, and null tags with regression tests. --- .../load/TsFileWriterManager.java | 16 ++ .../write/TsFilePrecalculatedChunkWriter.java | 14 ++ .../load/LoadTsFileTableSchemaTest.java | 181 ++++++++++++++++++ 3 files changed, 211 insertions(+) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java index 2c81646c53de3..fc8f38b850288 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/TsFileWriterManager.java @@ -24,12 +24,16 @@ import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex; import org.apache.iotdb.commons.file.SystemFileFactory; +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.commons.utils.RetryUtils; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.load.LoadFileException; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileConsensusOp; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -502,6 +506,18 @@ void prepare( fileLength)); } } + // Rebuild from the staged devices so recovered writers also retain the table schemas that + // Pipe needs to enumerate table data in the sealed file. + if (PathUtils.isTableModelDatabase(dataRegion.getDatabaseName())) { + for (final String tableName : writer.getTableNames()) { + final TsTable table = + DataNodeTableCache.getInstance() + .getTable(dataRegion.getDatabaseName(), tableName, false); + if (table != null) { + writer.registerTableSchema(TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table)); + } + } + } writer.close(); final TsFileResource tsFileResource = timePartition2Resource.get(entry.getKey()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java b/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java index a6c0d996c9817..b48b7d14cb17d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/tsfile/write/TsFilePrecalculatedChunkWriter.java @@ -31,6 +31,7 @@ import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.MeasurementMetadataIndexEntry; import org.apache.tsfile.file.metadata.MetadataIndexNode; +import org.apache.tsfile.file.metadata.TableSchema; import org.apache.tsfile.file.metadata.TimeseriesMetadata; import org.apache.tsfile.file.metadata.TsFileMetadata; import org.apache.tsfile.file.metadata.enums.MetadataIndexNodeType; @@ -53,6 +54,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.TreeMap; import java.util.stream.Collectors; @@ -80,6 +82,7 @@ public class TsFilePrecalculatedChunkWriter implements AutoCloseable { private final File file; private final Map>> device2MetadataMap = new TreeMap<>(); + private final Map tableSchemaMap = new TreeMap<>(); private long maxPhysicalFileSize; private boolean sealed; @@ -385,6 +388,7 @@ public void close() throws IOException { } tsFileMetadata.setTableMetadataIndexNodeMap(tableIndexNodeMap); + tsFileMetadata.setTableSchemaMap(tableSchemaMap); tsFileMetadata.setMetaOffset(metaOffset); tsFileMetadata.setBloomFilter(bloomFilter); @@ -446,6 +450,16 @@ public Map> getChunkMetadataListMap() { return result; } + public Set getTableNames() { + return device2MetadataMap.keySet().stream() + .map(IDeviceID::getTableName) + .collect(Collectors.toSet()); + } + + public void registerTableSchema(final TableSchema tableSchema) { + tableSchemaMap.put(tableSchema.getTableName(), tableSchema); + } + private String getFileForLog() { return file == null ? "" : file.getAbsolutePath(); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java new file mode 100644 index 0000000000000..306f3050c4882 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/LoadTsFileTableSchemaTest.java @@ -0,0 +1,181 @@ +/* + * 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.load; + +import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern; +import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.commons.schema.table.column.FieldColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TagColumnSchema; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent; +import org.apache.iotdb.db.pipe.event.common.tsfile.parser.table.TsFileInsertionEventTableParser; +import org.apache.iotdb.db.queryengine.plan.scheduler.load.ChunkOffsetCalculator; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.storageengine.dataregion.DataRegion; +import org.apache.iotdb.db.storageengine.load.splitter.ChunkData; +import org.apache.iotdb.db.storageengine.load.splitter.TsFileData; +import org.apache.iotdb.db.storageengine.load.splitter.TsFileSplitter; +import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.record.Tablet; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.Mockito; + +import java.io.File; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import static org.junit.Assert.assertEquals; + +public class LoadTsFileTableSchemaTest { + + private static final String DATABASE = "load_table_schema_test"; + private static final String TABLE = "measurements"; + private static final int ROW_COUNT = 20; + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private String[] originalLoadDirs; + private DataRegion dataRegion; + private TsTable table; + + @Before + public void setUp() throws Exception { + originalLoadDirs = IoTDBDescriptor.getInstance().getConfig().getLoadTsFileDirs(); + IoTDBDescriptor.getInstance() + .getConfig() + .setLoadTsFileDirs(new String[] {temporaryFolder.getRoot().getAbsolutePath()}); + dataRegion = Mockito.mock(DataRegion.class); + Mockito.when(dataRegion.getDatabaseName()).thenReturn(DATABASE); + Mockito.when(dataRegion.getDataRegionIdString()).thenReturn("0"); + Mockito.when(dataRegion.getNonSystemDatabaseName()).thenReturn(Optional.of(DATABASE)); + + table = new TsTable(TABLE); + table.addColumnSchema(new TagColumnSchema("device", TSDataType.STRING)); + table.addColumnSchema( + new FieldColumnSchema("value", TSDataType.INT32, TSEncoding.PLAIN, CompressionType.LZ4)); + DataNodeTableCache.getInstance().preUpdateTable(DATABASE, table, null); + DataNodeTableCache.getInstance().commitUpdateTable(DATABASE, TABLE, null); + } + + @After + public void tearDown() { + DataNodeTableCache.getInstance().invalid(DATABASE); + IoTDBDescriptor.getInstance().getConfig().setLoadTsFileDirs(originalLoadDirs); + } + + @Test + public void testLoadedTableRemainsReadableByPipe() throws Exception { + assertLoadedTableReadable(false, false); + } + + @Test + public void testRecoveredTableRemainsReadableByPipe() throws Exception { + assertLoadedTableReadable(true, false); + } + + @Test + public void testLoadedTableWithNullTagsRemainsReadableByPipe() throws Exception { + assertLoadedTableReadable(false, true); + } + + private void assertLoadedTableReadable(final boolean recover, final boolean nullTags) + throws Exception { + final File source = new File(temporaryFolder.getRoot(), "source.tsfile"); + final TableSchema schema = TsFileTableSchemaUtil.toTsFileTableSchemaNoAttribute(table); + try (final TsFileWriter writer = new TsFileWriter(source)) { + writer.registerTableSchema(schema); + final Tablet tablet = + new Tablet( + TABLE, + Arrays.asList("device", "value"), + Arrays.asList(TSDataType.STRING, TSDataType.INT32), + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD), + ROW_COUNT); + for (int i = 0; i < ROW_COUNT; i++) { + tablet.addTimestamp(i, i); + tablet.addValue("device", i, nullTags ? null : "d" + i % 2); + tablet.addValue("value", i, i); + } + writer.writeTable(tablet); + } + + final File taskDir = temporaryFolder.newFolder("task"); + TsFileWriterManager manager = new TsFileWriterManager(dataRegion, taskDir); + try { + final ChunkOffsetCalculator calculator = new ChunkOffsetCalculator(); + final List pieces = new ArrayList<>(); + new TsFileSplitter( + source, + data -> { + calculator.assign((ChunkData) data); + pieces.add(data); + return true; + }) + .splitTsFileByDataPartition(); + manager.writePiece(pieces); + if (recover) { + manager.close(true); + manager = new TsFileWriterManager(dataRegion, taskDir).recoverFromDisk(); + } + manager.prepare(false, Collections.emptyMap(), Collections.emptyMap()); + + final File staged = new File(taskDir, DATABASE + "-0-0.tsfile"); + int rowCount = 0; + try (final TsFileInsertionEventTableParser parser = + new TsFileInsertionEventTableParser( + staged, + new TablePattern(true, null, null), + Long.MIN_VALUE, + Long.MAX_VALUE, + null, + null, + null, + false)) { + for (final TabletInsertionEvent event : parser.toTabletInsertionEvents()) { + final PipeRawTabletInsertionEvent tabletEvent = (PipeRawTabletInsertionEvent) event; + rowCount += tabletEvent.convertToTablet().getRowSize(); + tabletEvent.clearReferenceCount(getClass().getName()); + } + } + assertEquals(ROW_COUNT, rowCount); + try (final TsFileSequenceReader reader = new TsFileSequenceReader(staged.getPath())) { + assertEquals(schema, reader.getTableSchemaMap().get(TABLE)); + } + } finally { + manager.close(); + } + } +}