Skip to content

Commit f3e1612

Browse files
authored
Fix downgraded TsFileResource serialization (#18155)
* Fix downgraded resource may not serialize normally * Revert FileIndex serialization
1 parent e231bbe commit f3e1612

3 files changed

Lines changed: 78 additions & 16 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/batch/IoTDBRegionMigrateNormalITForIoTV2BatchIT.java‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,13 +21,21 @@
2121

2222
import org.apache.iotdb.commons.utils.KillPoint.KillNode;
2323
import org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework;
24+
import org.apache.iotdb.it.env.EnvFactory;
2425
import org.apache.iotdb.it.framework.IoTDBTestRunner;
2526
import org.apache.iotdb.itbase.category.ClusterIT;
2627

28+
import org.junit.Assert;
2729
import org.junit.Test;
2830
import org.junit.experimental.categories.Category;
2931
import org.junit.runner.RunWith;
3032

33+
import java.sql.Connection;
34+
import java.sql.ResultSet;
35+
import java.sql.Statement;
36+
37+
import static org.apache.iotdb.util.MagicUtils.makeItCloseQuietly;
38+
3139
@Category({ClusterIT.class})
3240
@RunWith(IoTDBTestRunner.class)
3341
public class IoTDBRegionMigrateNormalITForIoTV2BatchIT
@@ -41,4 +49,30 @@ public void normal1C2DTest() throws Exception {
4149
public void normal3C3DTest() throws Exception {
4250
successTest(2, 3, 3, 3, noKillPoints(), noKillPoints(), KillNode.ALL_NODES);
4351
}
52+
53+
@Test
54+
public void migrateRegionWithDegradedTimeIndexTest() throws Exception {
55+
// set TsFileResource memory to 0 to trigger degrading
56+
EnvFactory.getEnv().getConfig().getCommonConfig().setQueryMemoryProportion("1:1:1:1:1:1:0");
57+
58+
successTest(1, 1, 1, 2, noKillPoints(), noKillPoints(), KillNode.ALL_NODES);
59+
60+
try (final Connection connection = makeItCloseQuietly(EnvFactory.getEnv().getConnection());
61+
final Statement statement = makeItCloseQuietly(connection.createStatement())) {
62+
assertCounts(statement, 1, 1);
63+
statement.execute("INSERT INTO root.sg.d1(timestamp,speed,temperature) values(101, 3, 4)");
64+
assertCounts(statement, 2, 2);
65+
}
66+
}
67+
68+
private static void assertCounts(
69+
final Statement statement, final long expectedSpeedCount, final long expectedTemperatureCount)
70+
throws Exception {
71+
try (final ResultSet resultSet =
72+
statement.executeQuery("select count(speed), count(temperature) from root.sg.d1")) {
73+
Assert.assertTrue(resultSet.next());
74+
Assert.assertEquals(expectedSpeedCount, resultSet.getLong(1));
75+
Assert.assertEquals(expectedTemperatureCount, resultSet.getLong(2));
76+
}
77+
}
4478
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java‎

Lines changed: 27 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -283,7 +283,7 @@ public synchronized void serialize() throws IOException {
283283

284284
private void serializeTo(BufferedOutputStream outputStream) throws IOException {
285285
ReadWriteIOUtils.write(VERSION_NUMBER, outputStream);
286-
timeIndex.serialize(outputStream);
286+
getTimeIndexForSerialization().serialize(outputStream);
287287

288288
ReadWriteIOUtils.write(maxPlanIndex, outputStream);
289289
ReadWriteIOUtils.write(minPlanIndex, outputStream);
@@ -317,6 +317,14 @@ private void serializeTo(BufferedOutputStream outputStream) throws IOException {
317317
ReadWriteIOUtils.write(isGeneratedByPipe, outputStream);
318318
}
319319

320+
private ITimeIndex getTimeIndexForSerialization() throws IOException {
321+
if (!(timeIndex instanceof FileTimeIndex) || !resourceFileExists()) {
322+
return timeIndex;
323+
}
324+
325+
return buildDeviceTimeIndex();
326+
}
327+
320328
/** deserialize from disk */
321329
public void deserialize() throws IOException {
322330
try (InputStream inputStream = fsFactory.getBufferedInputStream(file + RESOURCE_SUFFIX)) {
@@ -694,20 +702,13 @@ public ArrayDeviceTimeIndex buildDeviceTimeIndex(IDeviceID.Deserializer deserial
694702
throws IOException {
695703
readLock();
696704
try {
697-
if (!resourceFileExists()) {
698-
throw new IOException(StorageEngineMessages.RESOURCE_FILE_NOT_FOUND);
699-
}
700-
try (InputStream inputStream =
701-
FSFactoryProducer.getFSFactory()
702-
.getBufferedInputStream(file.getPath() + RESOURCE_SUFFIX)) {
703-
ReadWriteIOUtils.readByte(inputStream);
704-
ITimeIndex timeIndexFromResourceFile =
705-
ITimeIndex.createTimeIndex(inputStream, deserializer);
706-
if (!(timeIndexFromResourceFile instanceof ArrayDeviceTimeIndex)) {
707-
throw new IOException(
708-
StorageEngineMessages.CANNOT_BUILD_DEVICE_TIME_INDEX + file.getPath());
705+
try {
706+
ITimeIndex timeIndexFromResourceFile = deserializeTimeIndexFromResourceFile(deserializer);
707+
if (timeIndexFromResourceFile instanceof ArrayDeviceTimeIndex) {
708+
return (ArrayDeviceTimeIndex) timeIndexFromResourceFile;
709709
}
710-
return (ArrayDeviceTimeIndex) timeIndexFromResourceFile;
710+
throw new IOException(
711+
StorageEngineMessages.CANNOT_BUILD_DEVICE_TIME_INDEX + file.getPath());
711712
} catch (Exception e) {
712713
throw new IOException(
713714
String.format(
@@ -721,6 +722,18 @@ public ArrayDeviceTimeIndex buildDeviceTimeIndex(IDeviceID.Deserializer deserial
721722
}
722723
}
723724

725+
private ITimeIndex deserializeTimeIndexFromResourceFile(IDeviceID.Deserializer deserializer)
726+
throws IOException {
727+
if (!resourceFileExists()) {
728+
throw new IOException(StorageEngineMessages.RESOURCE_FILE_NOT_FOUND);
729+
}
730+
try (InputStream inputStream =
731+
FSFactoryProducer.getFSFactory().getBufferedInputStream(file.getPath() + RESOURCE_SUFFIX)) {
732+
ReadWriteIOUtils.readByte(inputStream);
733+
return ITimeIndex.createTimeIndex(inputStream, deserializer);
734+
}
735+
}
736+
724737
public ArrayDeviceTimeIndex buildDeviceTimeIndex() throws IOException {
725738
return buildDeviceTimeIndex(IDeviceID.Deserializer.DEFAULT_DESERIALIZER);
726739
}

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResourceTest.java‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,7 @@ public void tearDown() throws IOException {
9393
if (file.exists()) {
9494
FileUtils.delete(file);
9595
}
96-
File resourceFile = new File(file.getName() + TsFileResource.RESOURCE_SUFFIX);
96+
File resourceFile = new File(file.getPath() + TsFileResource.RESOURCE_SUFFIX);
9797
if (resourceFile.exists()) {
9898
FileUtils.delete(resourceFile);
9999
}
@@ -108,7 +108,22 @@ public void testSerializeAndDeserialize() throws IOException {
108108
}
109109

110110
@Test
111-
public void testDegradeAndFileTimeIndex() {
111+
public void testSerializeDegradedTimeIndex() throws IOException {
112+
tsFileResource.serialize();
113+
tsFileResource.degradeTimeIndex();
114+
115+
tsFileResource.serialize();
116+
117+
TsFileResource derTsFileResource = new TsFileResource(file);
118+
derTsFileResource.deserialize();
119+
Assert.assertEquals(
120+
ITimeIndex.ARRAY_DEVICE_TIME_INDEX_TYPE, derTsFileResource.getTimeIndexType());
121+
Assert.assertEquals(deviceToIndex.keySet(), derTsFileResource.getDevices());
122+
}
123+
124+
@Test
125+
public void testDegradeAndFileTimeIndex() throws IOException {
126+
tsFileResource.serialize();
112127
Assert.assertEquals(ITimeIndex.ARRAY_DEVICE_TIME_INDEX_TYPE, tsFileResource.getTimeIndexType());
113128
tsFileResource.degradeTimeIndex();
114129
Assert.assertEquals(ITimeIndex.FILE_TIME_INDEX_TYPE, tsFileResource.getTimeIndexType());

0 commit comments

Comments
 (0)