You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2023/10/11 04:01:29 UTC
[iotdb] 01/01: Fix bad tsfile being recoverd from wal
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch fix_aligned_file_recover
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 61b9a05377818c9974d96b3420c38317f3ed6bfd
Author: HTHou <hh...@outlook.com>
AuthorDate: Wed Oct 11 12:01:04 2023 +0800
Fix bad tsfile being recoverd from wal
---
.../wal/recover/file/TsFilePlanRedoer.java | 9 ++--
.../file/UnsealedTsFileRecoverPerformer.java | 61 +++++++++-------------
.../wal/recover/file/TsFilePlanRedoerTest.java | 35 ++++++++++---
.../iotdb/tsfile/file/header/ChunkGroupHeader.java | 3 ++
.../tsfile/read/TsFileSequenceReaderTest.java | 11 ++--
.../write/writer/RestorableTsFileIOWriterTest.java | 37 +++++++++++++
6 files changed, 102 insertions(+), 54 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java
index f862282d411..7bf3e136701 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java
@@ -78,12 +78,9 @@ public class TsFilePlanRedoer {
if (!node.hasValidMeasurements()) {
return;
}
+ String deviceId = node.getDevicePath().getFullPath();
if (tsFileResource != null) {
- String deviceId =
- node.isAligned()
- ? node.getDevicePath().getDevicePath().getFullPath()
- : node.getDevicePath().getFullPath();
- // orders of insert node is guaranteed by storage storageengine, just check time in the file
+ // orders of insert node is guaranteed by storage engine, just check time in the file
// the last chunk group may contain the same data with the logs, ignore such logs in seq file
long lastEndTime = tsFileResource.getEndTime(deviceId);
long minTimeInNode;
@@ -97,7 +94,7 @@ public class TsFilePlanRedoer {
}
}
- node.setDeviceID(DeviceIDFactory.getInstance().getDeviceID(node.getDevicePath()));
+ node.setDeviceID(DeviceIDFactory.getInstance().getDeviceID(deviceId));
if (node instanceof InsertRowNode) {
if (node.isAligned()) {
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 13a97b208af..9c883889143 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
@@ -36,7 +36,6 @@ 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;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
import org.slf4j.Logger;
@@ -119,44 +118,36 @@ public class UnsealedTsFileRecoverPerformer extends AbstractTsFileRecoverPerform
chunkMetadata.getMeasurementUid(), n -> new ArrayList<>());
list.add(chunkMetadata);
}
-
- for (List<ChunkMetadata> metadataList : measurementToChunkMetadatas.values()) {
- TSDataType dataType = metadataList.get(metadataList.size() - 1).getDataType();
- for (ChunkMetadata chunkMetaData : chunkMetadataList) {
- if (!chunkMetaData.getDataType().equals(dataType)) {
- continue;
- }
-
- // calculate startTime and endTime according to chunkMetaData and modifications
- long startTime = chunkMetaData.getStartTime();
- long endTime = chunkMetaData.getEndTime();
- long chunkHeaderOffset = chunkMetaData.getOffsetOfChunkHeader();
- if (modificationsForResource.containsKey(deviceId)
- && modificationsForResource
- .get(deviceId)
- .containsKey(chunkMetaData.getMeasurementUid())) {
- // exist deletion for current measurement
- for (Deletion modification :
- modificationsForResource.get(deviceId).get(chunkMetaData.getMeasurementUid())) {
- long fileOffset = modification.getFileOffset();
- if (chunkHeaderOffset < fileOffset) {
- // deletion is valid for current chunk
- long modsStartTime = modification.getStartTime();
- long modsEndTime = modification.getEndTime();
- if (startTime >= modsStartTime && endTime <= modsEndTime) {
- startTime = Long.MAX_VALUE;
- endTime = Long.MIN_VALUE;
- } else if (startTime >= modsStartTime && startTime <= modsEndTime) {
- startTime = modsEndTime + 1;
- } else if (endTime >= modsStartTime && endTime <= modsEndTime) {
- endTime = modsStartTime - 1;
- }
+ for (ChunkMetadata chunkMetaData : chunkMetadataList) {
+ // calculate startTime and endTime according to chunkMetaData and modifications
+ long startTime = chunkMetaData.getStartTime();
+ long endTime = chunkMetaData.getEndTime();
+ long chunkHeaderOffset = chunkMetaData.getOffsetOfChunkHeader();
+ if (modificationsForResource.containsKey(deviceId)
+ && modificationsForResource
+ .get(deviceId)
+ .containsKey(chunkMetaData.getMeasurementUid())) {
+ // exist deletion for current measurement
+ for (Deletion modification :
+ modificationsForResource.get(deviceId).get(chunkMetaData.getMeasurementUid())) {
+ long fileOffset = modification.getFileOffset();
+ if (chunkHeaderOffset < fileOffset) {
+ // deletion is valid for current chunk
+ long modsStartTime = modification.getStartTime();
+ long modsEndTime = modification.getEndTime();
+ if (startTime >= modsStartTime && endTime <= modsEndTime) {
+ startTime = Long.MAX_VALUE;
+ endTime = Long.MIN_VALUE;
+ } else if (startTime >= modsStartTime && startTime <= modsEndTime) {
+ startTime = modsEndTime + 1;
+ } else if (endTime >= modsStartTime && endTime <= modsEndTime) {
+ endTime = modsStartTime - 1;
}
}
}
- tsFileResource.updateStartTime(deviceId, startTime);
- tsFileResource.updateEndTime(deviceId, endTime);
}
+ tsFileResource.updateStartTime(deviceId, startTime);
+ tsFileResource.updateEndTime(deviceId, endTime);
}
}
tsFileResource.updatePlanIndexes(writer.getMinPlanIndex());
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java
index 5c440b16165..21d4c3dca5a 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java
@@ -187,27 +187,45 @@ public class TsFilePlanRedoerTest {
generateCompleteFile(file);
tsFileResource = new TsFileResource(file);
tsFileResource.updateStartTime(DEVICE3_NAME, 5);
- tsFileResource.updateStartTime(DEVICE3_NAME, 5);
+ tsFileResource.updateEndTime(DEVICE3_NAME, 5);
// generate InsertRowPlan
- long time = 6;
TSDataType[] dataTypes =
new TSDataType[] {
TSDataType.INT32, TSDataType.INT64, TSDataType.BOOLEAN, TSDataType.FLOAT, TSDataType.TEXT
};
Object[] columns = new Object[] {1, 1L, true, 1.0f, new Binary("1")};
- InsertRowNode insertRowNode =
+ InsertRowNode insertRowNode1 =
new InsertRowNode(
new PlanNodeId("0"),
new PartialPath(DEVICE3_NAME),
true,
new String[] {"s1", "s2", "s3", "s4", "s5"},
dataTypes,
- time,
+ 5,
columns,
false);
- insertRowNode.setMeasurementSchemas(
+ insertRowNode1.setMeasurementSchemas(
+ new MeasurementSchema[] {
+ new MeasurementSchema("s1", TSDataType.INT32),
+ new MeasurementSchema("s2", TSDataType.INT64),
+ new MeasurementSchema("s3", TSDataType.BOOLEAN),
+ new MeasurementSchema("s4", TSDataType.FLOAT),
+ new MeasurementSchema("s5", TSDataType.TEXT),
+ });
+
+ InsertRowNode insertRowNode2 =
+ new InsertRowNode(
+ new PlanNodeId("0"),
+ new PartialPath(DEVICE3_NAME),
+ true,
+ new String[] {"s1", "s2", "s3", "s4", "s5"},
+ dataTypes,
+ 6,
+ columns,
+ false);
+ insertRowNode2.setMeasurementSchemas(
new MeasurementSchema[] {
new MeasurementSchema("s1", TSDataType.INT32),
new MeasurementSchema("s2", TSDataType.INT64),
@@ -218,7 +236,8 @@ public class TsFilePlanRedoerTest {
// redo InsertTabletPlan, vsg processor is used to test IdTable, don't test IdTable here
TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource, true);
- planRedoer.redoInsert(insertRowNode);
+ planRedoer.redoInsert(insertRowNode1);
+ planRedoer.redoInsert(insertRowNode2);
// check data in memTable
IMemTable recoveryMemTable = planRedoer.getRecoveryMemTable();
@@ -235,13 +254,13 @@ public class TsFilePlanRedoerTest {
new MeasurementSchema("s5", TSDataType.TEXT, TSEncoding.PLAIN)));
ReadOnlyMemChunk memChunk = recoveryMemTable.query(fullPath, Long.MIN_VALUE, null);
IPointReader iterator = memChunk.getPointReader();
- time = 6;
+ int time = 6;
while (iterator.hasNextTimeValuePair()) {
TimeValuePair timeValuePair = iterator.nextTimeValuePair();
assertEquals(time, timeValuePair.getTimestamp());
assertEquals(1, timeValuePair.getValue().getVector()[0].getInt());
assertEquals(1L, timeValuePair.getValue().getVector()[1].getLong());
- assertEquals(true, timeValuePair.getValue().getVector()[2].getBoolean());
+ assertTrue(timeValuePair.getValue().getVector()[2].getBoolean());
assertEquals(1, timeValuePair.getValue().getVector()[3].getFloat(), 0.00001);
assertEquals(Binary.valueOf("1"), timeValuePair.getValue().getVector()[4].getBinary());
++time;
diff --git a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/ChunkGroupHeader.java b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/ChunkGroupHeader.java
index c68ea27e9e8..406b1dfe9c4 100644
--- a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/ChunkGroupHeader.java
+++ b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/ChunkGroupHeader.java
@@ -73,6 +73,9 @@ public class ChunkGroupHeader {
}
String deviceID = ReadWriteIOUtils.readVarIntString(inputStream);
+ if (deviceID == null || deviceID.isEmpty()) {
+ throw new IOException("DeviceId is empty");
+ }
return new ChunkGroupHeader(deviceID);
}
diff --git a/iotdb-core/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java b/iotdb-core/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java
index a69eec5bfbd..a94b08054cc 100644
--- a/iotdb-core/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java
+++ b/iotdb-core/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java
@@ -125,7 +125,7 @@ public class TsFileSequenceReaderTest {
List<ChunkMetadata> metadataList = chunkMetadataMap.get("s" + id);
int numOfPoints = 0;
for (ChunkMetadata metadata : metadataList) {
- numOfPoints += metadata.getNumOfPoints();
+ numOfPoints += (int) metadata.getNumOfPoints();
}
Assert.assertEquals(res[i], numOfPoints);
}
@@ -167,9 +167,10 @@ public class TsFileSequenceReaderTest {
}
// read tsfile with selfCheck method
- TsFileSequenceReader reader = new TsFileSequenceReader(FILE_PATH);
- Assert.assertEquals(
- TsFileCheckStatus.COMPLETE_FILE,
- reader.selfCheck(new HashMap<>(), new ArrayList<>(), false));
+ try (TsFileSequenceReader reader = new TsFileSequenceReader(FILE_PATH)) {
+ Assert.assertEquals(
+ TsFileCheckStatus.COMPLETE_FILE,
+ reader.selfCheck(new HashMap<>(), new ArrayList<>(), false));
+ }
}
}
diff --git a/iotdb-core/tsfile/src/test/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriterTest.java b/iotdb-core/tsfile/src/test/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriterTest.java
index b0a67f82022..6033f17fe25 100644
--- a/iotdb-core/tsfile/src/test/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriterTest.java
+++ b/iotdb-core/tsfile/src/test/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriterTest.java
@@ -253,6 +253,43 @@ public class RestorableTsFileIOWriterTest {
writer = new TsFileWriter(rWriter);
writer.close();
assertNotEquals(TsFileIOWriter.MAGIC_STRING_BYTES.length, rWriter.getTruncatedSize());
+ assertEquals(89, rWriter.getTruncatedSize());
+ rWriter.close();
+
+ TsFileSequenceReader reader = new TsFileSequenceReader(FILE_NAME);
+ List<ChunkMetadata> chunkMetadataList = reader.getChunkMetadataList(new Path("d1", "s1", true));
+ assertNotNull(chunkMetadataList);
+ chunkMetadataList = reader.getChunkMetadataList(new Path("d1", "s2", true));
+ assertNotNull(chunkMetadataList);
+ reader.close();
+ }
+
+ @Test
+ public void testAChunkGroupEndWithALotOfZeroBytes() throws Exception {
+ TsFileWriter writer = new TsFileWriter(file);
+ writer.registerTimeseries(
+ new Path("d1"), new MeasurementSchema("s1", TSDataType.FLOAT, TSEncoding.RLE));
+ writer.registerTimeseries(
+ new Path("d1"), new MeasurementSchema("s2", TSDataType.FLOAT, TSEncoding.RLE));
+ writer.write(
+ new TSRecord(1, "d1")
+ .addTuple(new FloatDataPoint("s1", 5))
+ .addTuple(new FloatDataPoint("s2", 4)));
+ writer.write(
+ new TSRecord(2, "d1")
+ .addTuple(new FloatDataPoint("s1", 5))
+ .addTuple(new FloatDataPoint("s2", 4)));
+ writer.flushAllChunkGroups();
+ // write 10 Zero bytes
+ for (int i = 0; i < 10; i++) {
+ writer.getIOWriter().writeChunkGroupMarkerForTest();
+ }
+ writer.getIOWriter().close();
+ RestorableTsFileIOWriter rWriter = new RestorableTsFileIOWriter(file);
+ writer = new TsFileWriter(rWriter);
+ writer.close();
+ assertNotEquals(TsFileIOWriter.MAGIC_STRING_BYTES.length, rWriter.getTruncatedSize());
+ assertEquals(89, rWriter.getTruncatedSize());
rWriter.close();
TsFileSequenceReader reader = new TsFileSequenceReader(FILE_NAME);