This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch fix_serialization_of_downgraded_resource in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ef0d353c7114ec2bf849b0b96d419a3eae5beed6 Author: Tian Jiang <[email protected]> AuthorDate: Wed Jul 8 17:58:42 2026 +0800 Fix downgraded resource may not serialize normally --- .../IoTDBRegionMigrateNormalITForIoTV2BatchIT.java | 34 ++++++++++++++++ .../dataregion/tsfile/TsFileResource.java | 46 +++++++++++++++------- .../dataregion/tsfile/timeindex/FileTimeIndex.java | 12 ++++-- .../dataregion/tsfile/TsFileResourceTest.java | 33 +++++++++++++++- 4 files changed, 106 insertions(+), 19 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/batch/IoTDBRegionMigrateNormalITForIoTV2BatchIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/batch/IoTDBRegionMigrateNormalITForIoTV2BatchIT.java index b763b424002..23cd7236ef2 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/batch/IoTDBRegionMigrateNormalITForIoTV2BatchIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/batch/IoTDBRegionMigrateNormalITForIoTV2BatchIT.java @@ -21,13 +21,21 @@ package org.apache.iotdb.confignode.it.regionmigration.pass.commit.batch; import org.apache.iotdb.commons.utils.KillPoint.KillNode; import org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework; +import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.ClusterIT; +import org.junit.Assert; import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; + +import static org.apache.iotdb.util.MagicUtils.makeItCloseQuietly; + @Category({ClusterIT.class}) @RunWith(IoTDBTestRunner.class) public class IoTDBRegionMigrateNormalITForIoTV2BatchIT @@ -41,4 +49,30 @@ public class IoTDBRegionMigrateNormalITForIoTV2BatchIT public void normal3C3DTest() throws Exception { successTest(2, 3, 3, 3, noKillPoints(), noKillPoints(), KillNode.ALL_NODES); } + + @Test + public void migrateRegionWithDegradedTimeIndexTest() throws Exception { + // set TsFileResource memory to 0 to trigger degrading + EnvFactory.getEnv().getConfig().getCommonConfig().setQueryMemoryProportion("1:1:1:1:1:1:0"); + + successTest(1, 1, 1, 2, noKillPoints(), noKillPoints(), KillNode.ALL_NODES); + + try (final Connection connection = makeItCloseQuietly(EnvFactory.getEnv().getConnection()); + final Statement statement = makeItCloseQuietly(connection.createStatement())) { + assertCounts(statement, 1, 1); + statement.execute("INSERT INTO root.sg.d1(timestamp,speed,temperature) values(101, 3, 4)"); + assertCounts(statement, 2, 2); + } + } + + private static void assertCounts( + final Statement statement, final long expectedSpeedCount, final long expectedTemperatureCount) + throws Exception { + try (final ResultSet resultSet = + statement.executeQuery("select count(speed), count(temperature) from root.sg.d1")) { + Assert.assertTrue(resultSet.next()); + Assert.assertEquals(expectedSpeedCount, resultSet.getLong(1)); + Assert.assertEquals(expectedTemperatureCount, resultSet.getLong(2)); + } + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java index 80476e465ec..ccc37a84ce3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java @@ -283,7 +283,7 @@ public class TsFileResource implements PersistentResource, Cloneable { private void serializeTo(BufferedOutputStream outputStream) throws IOException { ReadWriteIOUtils.write(VERSION_NUMBER, outputStream); - timeIndex.serialize(outputStream); + getTimeIndexForSerialization().serialize(outputStream); ReadWriteIOUtils.write(maxPlanIndex, outputStream); ReadWriteIOUtils.write(minPlanIndex, outputStream); @@ -317,6 +317,15 @@ public class TsFileResource implements PersistentResource, Cloneable { ReadWriteIOUtils.write(isGeneratedByPipe, outputStream); } + private ITimeIndex getTimeIndexForSerialization() throws IOException { + if (!(timeIndex instanceof FileTimeIndex) || !resourceFileExists()) { + return timeIndex; + } + + ITimeIndex timeIndexFromResourceFile = deserializeTimeIndexFromResourceFile(); + return timeIndexFromResourceFile instanceof FileTimeIndex ? timeIndex : timeIndexFromResourceFile; + } + /** deserialize from disk */ public void deserialize() throws IOException { try (InputStream inputStream = fsFactory.getBufferedInputStream(file + RESOURCE_SUFFIX)) { @@ -694,20 +703,13 @@ public class TsFileResource implements PersistentResource, Cloneable { throws IOException { readLock(); try { - if (!resourceFileExists()) { - throw new IOException(StorageEngineMessages.RESOURCE_FILE_NOT_FOUND); - } - try (InputStream inputStream = - FSFactoryProducer.getFSFactory() - .getBufferedInputStream(file.getPath() + RESOURCE_SUFFIX)) { - ReadWriteIOUtils.readByte(inputStream); - ITimeIndex timeIndexFromResourceFile = - ITimeIndex.createTimeIndex(inputStream, deserializer); - if (!(timeIndexFromResourceFile instanceof ArrayDeviceTimeIndex)) { - throw new IOException( - StorageEngineMessages.CANNOT_BUILD_DEVICE_TIME_INDEX + file.getPath()); + try { + ITimeIndex timeIndexFromResourceFile = deserializeTimeIndexFromResourceFile(deserializer); + if (timeIndexFromResourceFile instanceof ArrayDeviceTimeIndex) { + return (ArrayDeviceTimeIndex) timeIndexFromResourceFile; } - return (ArrayDeviceTimeIndex) timeIndexFromResourceFile; + throw new IOException( + StorageEngineMessages.CANNOT_BUILD_DEVICE_TIME_INDEX + file.getPath()); } catch (Exception e) { throw new IOException( String.format( @@ -721,6 +723,22 @@ public class TsFileResource implements PersistentResource, Cloneable { } } + private ITimeIndex deserializeTimeIndexFromResourceFile() throws IOException { + return deserializeTimeIndexFromResourceFile(IDeviceID.Deserializer.DEFAULT_DESERIALIZER); + } + + private ITimeIndex deserializeTimeIndexFromResourceFile(IDeviceID.Deserializer deserializer) + throws IOException { + if (!resourceFileExists()) { + throw new IOException(StorageEngineMessages.RESOURCE_FILE_NOT_FOUND); + } + try (InputStream inputStream = + FSFactoryProducer.getFSFactory().getBufferedInputStream(file.getPath() + RESOURCE_SUFFIX)) { + ReadWriteIOUtils.readByte(inputStream); + return ITimeIndex.createTimeIndex(inputStream, deserializer); + } + } + public ArrayDeviceTimeIndex buildDeviceTimeIndex() throws IOException { return buildDeviceTimeIndex(IDeviceID.Deserializer.DEFAULT_DESERIALIZER); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/timeindex/FileTimeIndex.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/timeindex/FileTimeIndex.java index 0230c052392..0944611f729 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/timeindex/FileTimeIndex.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/timeindex/FileTimeIndex.java @@ -71,18 +71,24 @@ public class FileTimeIndex implements ITimeIndex { @Override public void serialize(OutputStream outputStream) throws IOException { - throw new UnsupportedOperationException(); + ReadWriteIOUtils.write(FILE_TIME_INDEX_TYPE, outputStream); + ReadWriteIOUtils.write(startTime, outputStream); + ReadWriteIOUtils.write(endTime, outputStream); } @Override public FileTimeIndex deserialize(InputStream inputStream, IDeviceID.Deserializer deserializer) throws IOException { - throw new UnsupportedOperationException(); + startTime = ReadWriteIOUtils.readLong(inputStream); + endTime = ReadWriteIOUtils.readLong(inputStream); + return this; } @Override public FileTimeIndex deserialize(ByteBuffer buffer) { - throw new UnsupportedOperationException(); + startTime = buffer.getLong(); + endTime = buffer.getLong(); + return this; } @Override diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResourceTest.java index d637450eee3..f5287707f16 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResourceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResourceTest.java @@ -28,6 +28,7 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.v1.Modification import org.apache.iotdb.db.storageengine.dataregion.modification.v1.ModificationFileV1; import org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ArrayDeviceTimeIndex; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex; import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex; import org.apache.iotdb.db.utils.constant.TestConstant; @@ -93,7 +94,7 @@ public class TsFileResourceTest { if (file.exists()) { FileUtils.delete(file); } - File resourceFile = new File(file.getName() + TsFileResource.RESOURCE_SUFFIX); + File resourceFile = new File(file.getPath() + TsFileResource.RESOURCE_SUFFIX); if (resourceFile.exists()) { FileUtils.delete(resourceFile); } @@ -108,7 +109,35 @@ public class TsFileResourceTest { } @Test - public void testDegradeAndFileTimeIndex() { + public void testSerializeDegradedTimeIndex() throws IOException { + tsFileResource.serialize(); + tsFileResource.degradeTimeIndex(); + + tsFileResource.serialize(); + + TsFileResource derTsFileResource = new TsFileResource(file); + derTsFileResource.deserialize(); + Assert.assertEquals( + ITimeIndex.ARRAY_DEVICE_TIME_INDEX_TYPE, derTsFileResource.getTimeIndexType()); + Assert.assertEquals(deviceToIndex.keySet(), derTsFileResource.getDevices()); + } + + @Test + public void testSerializeFileTimeIndex() throws IOException { + tsFileResource.setTimeIndex(new FileTimeIndex(1, 2)); + + tsFileResource.serialize(); + + TsFileResource derTsFileResource = new TsFileResource(file); + derTsFileResource.deserialize(); + Assert.assertEquals(ITimeIndex.FILE_TIME_INDEX_TYPE, derTsFileResource.getTimeIndexType()); + Assert.assertEquals(1, derTsFileResource.getFileStartTime()); + Assert.assertEquals(2, derTsFileResource.getFileEndTime()); + } + + @Test + public void testDegradeAndFileTimeIndex() throws IOException { + tsFileResource.serialize(); Assert.assertEquals(ITimeIndex.ARRAY_DEVICE_TIME_INDEX_TYPE, tsFileResource.getTimeIndexType()); tsFileResource.degradeTimeIndex(); Assert.assertEquals(ITimeIndex.FILE_TIME_INDEX_TYPE, tsFileResource.getTimeIndexType());
