This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new ab15c0a19b0 Recover WAL metadata from readable entries and quarantine
unrecoverable files (#18693)
ab15c0a19b0 is described below
commit ab15c0a19b0f7869300d3650045383e01c02d33d
Author: Jiang Tian <[email protected]>
AuthorDate: Thu Sep 24 14:20:55 2026 +0800
Recover WAL metadata from readable entries and quarantine unrecoverable
files (#18693)
* Recover WAL metadata from readable entries and quarantine unrecoverable
files
* Guard WAL recovery against oversized compressed segments
* check wal entry size in WALByteBufReader
* spotless
---
.../iotdb/db/i18n/StorageEngineMessages.java | 2 +
.../iotdb/db/i18n/StorageEngineMessages.java | 2 +
.../dataregion/wal/buffer/WALBuffer.java | 11 +-
.../dataregion/wal/io/WALByteBufReader.java | 15 +-
.../dataregion/wal/io/WALFileVersion.java | 19 ++
.../dataregion/wal/io/WALInputStream.java | 82 ++++++-
.../dataregion/wal/io/WALMetaData.java | 98 +++++++--
.../storageengine/dataregion/wal/io/WALReader.java | 12 +-
.../storageengine/dataregion/wal/io/WALWriter.java | 17 +-
.../storageengine/dataregion/wal/node/WALNode.java | 5 +-
.../dataregion/wal/recover/WALNodeRecoverTask.java | 54 ++++-
.../dataregion/wal/recover/WALRepairWriter.java | 125 +++++++++--
.../dataregion/wal/io/WALFileTest.java | 240 +++++++++++++++++++--
.../wal/recover/WALRepairWriterTest.java | 140 +++++++++---
14 files changed, 709 insertions(+), 113 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
index 5d946aed646..12ef8456800 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java
@@ -348,6 +348,8 @@ public final class StorageEngineMessages {
public static final String FAIL_TO_CREATE_WAL_NODE_DISKS_FULL = "Fail to
create wal node because all disks of wal folders are full.";
public static final String FAILED_TO_CREATE_WAL_NODE_AFTER_RETRIES = "Failed
to create WAL node after retries for identifier: ";
public static final String FAIL_TO_CREATE_WAL_NODE = "Fail to create wal
node";
+ public static final String OVER_SIZED_WAL_ENTRY = "The wal entry size %d
exceeds the limit %d, which may be a result of file corruption or configuration
change."
+ + "Please increase wal_buffer_size_in_byte or quarantine the file %s";
// ======================== Flush ========================
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
index 8324777fc85..a69ae2a6c8e 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java
@@ -348,6 +348,8 @@ public final class StorageEngineMessages {
public static final String FAIL_TO_CREATE_WAL_NODE_DISKS_FULL = "由于 WAL
目录的所有磁盘已满，无法创建 WAL 节点。";
public static final String FAILED_TO_CREATE_WAL_NODE_AFTER_RETRIES =
"重试后仍无法创建 WAL 节点，标识符: ";
public static final String FAIL_TO_CREATE_WAL_NODE = "创建 WAL 节点失败";
+ public static final String OVER_SIZED_WAL_ENTRY = "WAL条目大小 %d 超过阈值 %d,
可能是由于文件损坏或者系统配置变更."
+ + "请提高 wal_buffer_size_in_byte 或者隔离该文件 %s";
// ======================== Flush ========================
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
index 5cee2ff2675..27e36f7574c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
@@ -53,7 +53,6 @@ import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -892,9 +891,9 @@ public class WALBuffer extends AbstractWALBuffer {
id -> {
try {
File file = WALFileUtils.getWALFile(new File(logDirectory), id);
- return WALMetaData.readFromWALFile(
- file, FileChannel.open(file.toPath(),
StandardOpenOption.READ))
- .getMemTablesId();
+ try (FileChannel channel = FileChannel.open(file.toPath(),
StandardOpenOption.READ)) {
+ return WALMetaData.readFromWALFile(file,
channel).getMemTablesId();
+ }
} catch (BrokenWALFileException e) {
logger.warn(
StorageEngineMessages
@@ -911,7 +910,9 @@ public class WALBuffer extends AbstractWALBuffer {
e);
DataNodeExceptionMetrics.getInstance().recordSuspiciousDiskException(e);
}
- return Collections.emptySet();
+ // An unreadable WAL may still contain memTables. Treat the ids as
unknown so callers
+ // retain the file instead of deleting it as if it were an empty WAL.
+ return null;
});
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
index 70973e014c3..3fa8f23a2ce 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.io;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
+import org.apache.iotdb.db.i18n.StorageEngineMessages;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
import java.io.Closeable;
@@ -30,6 +31,8 @@ import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.List;
+import static
org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALBuffer.ONE_THIRD_WAL_BUFFER_SIZE;
+
/**
* This reader returns {@link WALEntry} as {@link ByteBuffer}, the usage of
WALByteBufReader is like
* {@link Iterator}.
@@ -56,7 +59,9 @@ public class WALByteBufReader implements Closeable {
}
public WALByteBufReader(File logFile, WALMetaData metaDataSnapshot) throws
IOException {
- WALInputStream walInputStream = new WALInputStream(logFile);
+ // A snapshot supplies the entry boundary for active files and recovered
prefixes, whose footer
+ // may be absent or damaged.
+ WALInputStream walInputStream = new WALInputStream(logFile, true);
try {
this.walInputStream = walInputStream;
this.logStream = new DataInputStream(walInputStream);
@@ -81,6 +86,14 @@ public class WALByteBufReader implements Closeable {
public ByteBuffer next() throws IOException {
currentEntryIndex++;
int size = sizeIterator.next();
+ if (size > ONE_THIRD_WAL_BUFFER_SIZE) {
+ throw new IOException(
+ String.format(
+ StorageEngineMessages.OVER_SIZED_WAL_ENTRY,
+ size,
+ ONE_THIRD_WAL_BUFFER_SIZE,
+ walInputStream.logFile));
+ }
// TODO: Reuse this buffer
ByteBuffer buffer = ByteBuffer.allocate(size);
/*
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileVersion.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileVersion.java
index 32cdc535ba6..f643c78e8dd 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileVersion.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileVersion.java
@@ -79,4 +79,23 @@ public enum WALFileVersion {
channel.position(originalPosition);
}
}
+
+ /**
+ * Returns whether the channel contains no bytes or only the V2/V3 header
magic.
+ *
+ * <p>A WAL writer creates the header before the first entry is available
and deliberately leaves
+ * that header-only file in place when it is closed. Such a file has no
metadata trailer to read,
+ * but it is still a valid empty WAL file.
+ */
+ public static boolean isEmptyOrHeaderOnly(FileChannel channel) throws
IOException {
+ long size = channel.size();
+ if (size == 0) {
+ return true;
+ }
+ if (size != V2.versionBytes.length && size != V3.versionBytes.length) {
+ return false;
+ }
+ WALFileVersion version = getVersion(channel);
+ return version == V2 || version == V3;
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
index 61e62b3e734..1c20982be67 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
@@ -22,8 +22,10 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.utils.IOUtils;
import org.apache.iotdb.db.i18n.StorageEngineMessages;
import org.apache.iotdb.db.service.metrics.WritingMetrics;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALBuffer;
import org.apache.iotdb.db.utils.MmapUtil;
+import org.apache.tsfile.compress.ICompressor;
import org.apache.tsfile.compress.IUnCompressor;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.slf4j.Logger;
@@ -59,16 +61,30 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
Aka, the last byte of the last segment.
*/
private long endOffset = -1;
+ private long logicalReadOffset;
+ private boolean recoveringEntries;
WALFileVersion version;
public WALInputStream(File logFile) throws IOException {
+ this(logFile, false);
+ }
+
+ WALInputStream(File logFile, boolean ignoreMetadata) throws IOException {
+ recoveringEntries = ignoreMetadata;
channel = FileChannel.open(logFile.toPath());
this.logFile = logFile;
try {
fileSize = channel.size();
analyzeFileVersion();
- getEndOffset();
+ if (ignoreMetadata) {
+ // Recovery must not trust even a plausible footer length. Stop at the
entry end marker
+ // or the first unreadable entry instead.
+ endOffset = fileSize;
+ channel.position(version == WALFileVersion.V1 ? 0 :
version.getVersionBytes().length);
+ } else {
+ getEndOffset();
+ }
} catch (Exception e) {
channel.close();
throw e;
@@ -77,8 +93,9 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
private void getEndOffset() throws IOException {
try {
- if (channel.size() < WALFileVersion.V2.getVersionBytes().length +
Integer.BYTES) {
- // An broken file
+ if (WALFileVersion.isEmptyOrHeaderOnly(channel)
+ || channel.size() < WALFileVersion.V2.getVersionBytes().length +
Integer.BYTES) {
+ // Empty and incomplete files have no metadata trailer, so there is no
segment to read.
endOffset = channel.size();
return;
}
@@ -124,7 +141,18 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
IOUtils.readFully(channel, metadataSizeBuf, position);
metadataSizeBuf.flip();
int metadataSize = metadataSizeBuf.getInt();
- endOffset = channel.size() - version.getVersionBytes().length -
Integer.BYTES - metadataSize;
+ long dataStart =
+ version == WALFileVersion.V2 || version == WALFileVersion.V3
+ ? version.getVersionBytes().length
+ : 0;
+ long dataEnd = position - metadataSize;
+ if (metadataSize < 0 || dataEnd < dataStart || dataEnd > position) {
+ // A damaged metadata length must not make the reader skip valid
entries or seek before the
+ // file header. Scan the remaining bytes so recovery can retain any
readable prefix.
+ endOffset = channel.size();
+ } else {
+ endOffset = dataEnd;
+ }
} finally {
if (version == WALFileVersion.V2 || version == WALFileVersion.V3) {
// Set the position back to the end of head magic string
@@ -145,7 +173,9 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
if (Objects.isNull(dataBuffer) || dataBuffer.position() >=
dataBuffer.limit()) {
loadNextSegment();
}
- return dataBuffer.get() & 0xFF;
+ int value = dataBuffer.get() & 0xFF;
+ logicalReadOffset++;
+ return value;
}
@Override
@@ -155,6 +185,7 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
}
if (dataBuffer.remaining() >= len) {
dataBuffer.get(b, off, len);
+ logicalReadOffset += len;
return len;
}
int toBeRead = len;
@@ -162,6 +193,7 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
int remaining = dataBuffer.remaining();
int bytesRead = Math.min(remaining, toBeRead);
dataBuffer.get(b, off, bytesRead);
+ logicalReadOffset += bytesRead;
off += bytesRead;
toBeRead -= bytesRead;
if (toBeRead > 0) {
@@ -223,6 +255,28 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
private void loadNextSegmentV2() throws IOException {
long position = channel.position();
SegmentInfo segmentInfo = getNextSegmentInfo();
+ long remainingBytes = fileSize - channel.position();
+ if (recoveringEntries && segmentInfo.compressionType ==
CompressionType.UNCOMPRESSED) {
+ // Complete entries in a partially written uncompressed segment remain
readable. A compressed
+ // segment needs its full payload before any of its entries can be
recovered.
+ segmentInfo.dataInDiskSize = (int) Math.min(segmentInfo.dataInDiskSize,
remainingBytes);
+ segmentInfo.uncompressedSize = segmentInfo.dataInDiskSize;
+ }
+ if (segmentInfo.dataInDiskSize <= 0
+ || segmentInfo.uncompressedSize <= 0
+ || segmentInfo.dataInDiskSize > remainingBytes) {
+ throw new EOFException(StorageEngineMessages.UNEXPECTED_END_OF_FILE);
+ }
+ // Recovery inspects untrusted headers. Bound allocations by the writer's
configured segment
+ // capacity so a tiny corrupt payload cannot request a huge decompression
buffer.
+ if (recoveringEntries
+ && (segmentInfo.uncompressedSize > WALBuffer.ONE_THIRD_WAL_BUFFER_SIZE
+ || (segmentInfo.compressionType != CompressionType.UNCOMPRESSED
+ && segmentInfo.dataInDiskSize
+ > ICompressor.getCompressor(segmentInfo.compressionType)
+
.getMaxBytesForCompression(WALBuffer.ONE_THIRD_WAL_BUFFER_SIZE)))) {
+ throw new EOFException(StorageEngineMessages.UNEXPECTED_END_OF_FILE);
+ }
try {
if (segmentInfo.compressionType != CompressionType.UNCOMPRESSED) {
// A compressed segment
@@ -352,10 +406,20 @@ public class WALInputStream extends InputStream
implements AutoCloseable {
public WALMetaData getWALMetaData() throws IOException {
long position = channel.position();
- channel.position(0);
- WALMetaData walMetaData = WALMetaData.readFromWALFile(logFile, channel);
- channel.position(position);
- return walMetaData;
+ try {
+ WALMetaData walMetaData = WALMetaData.readFromWALFile(logFile, channel);
+ if (walMetaData.isRecoveredFromEntries()) {
+ endOffset = fileSize;
+ recoveringEntries = true;
+ }
+ return walMetaData;
+ } finally {
+ channel.position(position);
+ }
+ }
+
+ public long getLogicalReadOffset() {
+ return logicalReadOffset;
}
private SegmentInfo getNextSegmentInfo() throws IOException {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java
index b0177325fda..50ae0961ce1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java
@@ -22,7 +22,10 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.io;
import org.apache.iotdb.commons.utils.IOUtils;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
import org.apache.iotdb.db.i18n.StorageEngineMessages;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
import
org.apache.iotdb.db.storageengine.dataregion.wal.exception.BrokenWALFileException;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRepairWriter;
import org.apache.iotdb.db.utils.SerializedSize;
import org.slf4j.Logger;
@@ -60,6 +63,7 @@ public class WALMetaData implements SerializedSize {
// memTable ids of this wal file
private final Set<Long> memTablesId;
private long truncateOffSet = 0;
+ private boolean recoveredFromEntries;
// V3 fields: file-level data timestamp range for timestamp-based seek
private long minDataTs = Long.MAX_VALUE;
@@ -218,9 +222,17 @@ public class WALMetaData implements SerializedSize {
public static WALMetaData deserialize(ByteBuffer buffer, WALFileVersion
version) {
long firstSearchIndex = buffer.getLong();
int entriesNum = buffer.getInt();
+ // Reject corrupted counts before allocating arrays from an untrusted
footer.
+ if (entriesNum < 0 || entriesNum > buffer.remaining() / Integer.BYTES) {
+ throw new
IllegalArgumentException(StorageEngineMessages.UNEXPECTED_EXCEPTION);
+ }
List<Integer> buffersSize = new ArrayList<>(entriesNum);
for (int i = 0; i < entriesNum; ++i) {
- buffersSize.add(buffer.getInt());
+ int size = buffer.getInt();
+ if (size <= 0) {
+ throw new
IllegalArgumentException(StorageEngineMessages.UNEXPECTED_EXCEPTION);
+ }
+ buffersSize.add(size);
}
Set<Long> memTablesId = new HashSet<>();
final boolean serializedEmptyV3WithoutMemTableCount =
@@ -229,6 +241,9 @@ public class WALMetaData implements SerializedSize {
&& buffer.remaining() ==
V3_EMPTY_METADATA_REMAINING_WITHOUT_MEMTABLE_COUNT;
if (buffer.hasRemaining() && !serializedEmptyV3WithoutMemTableCount) {
int memTablesIdNum = buffer.getInt();
+ if (memTablesIdNum < 0 || memTablesIdNum > buffer.remaining() /
Long.BYTES) {
+ throw new
IllegalArgumentException(StorageEngineMessages.UNEXPECTED_EXCEPTION);
+ }
for (int i = 0; i < memTablesIdNum; ++i) {
memTablesId.add(buffer.getLong());
}
@@ -244,6 +259,11 @@ public class WALMetaData implements SerializedSize {
}
final short defaultNodeId = buffer.getShort();
final int overrideCount = buffer.getInt();
+ if (overrideCount < 0
+ || overrideCount > entriesNum
+ || overrideCount > buffer.remaining() / (Integer.BYTES +
Short.BYTES)) {
+ throw new
IllegalArgumentException(StorageEngineMessages.UNEXPECTED_EXCEPTION);
+ }
final int[] overrideIndexes = new int[overrideCount];
final short[] overrideNodeIds = new short[overrideCount];
for (int i = 0; i < overrideCount; i++) {
@@ -324,6 +344,7 @@ public class WALMetaData implements SerializedSize {
WALMetaData copy =
new WALMetaData(firstSearchIndex, new ArrayList<>(buffersSize), new
HashSet<>(memTablesId));
copy.truncateOffSet = truncateOffSet;
+ copy.recoveredFromEntries = recoveredFromEntries;
copy.physicalTimes.addAll(physicalTimes);
copy.nodeIds.addAll(nodeIds);
copy.localSeqs.addAll(localSeqs);
@@ -352,8 +373,56 @@ public class WALMetaData implements SerializedSize {
}
public static WALMetaData readFromWALFile(File logFile, FileChannel channel)
throws IOException {
- if (channel.size() < WALFileVersion.V2.getVersionBytes().length
- || !isValidMagicString(channel)) {
+ try {
+ return readFromWALFileWithoutRecovery(logFile, channel);
+ } catch (IOException metadataFailure) {
+ logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_LOGS_SKIP, logFile,
metadataFailure);
+ // Keep the original file when some entries are readable. The
reconstructed metadata is an
+ // in-memory view of that prefix; rewriting compressed segments while
readers hold the file
+ // open would invalidate their offsets.
+ // Writer progress that exists only in a V3 footer cannot be recovered
from entry bodies;
+ // add() supplies the existing unknown/default progress values for those
fields.
+ WALMetaData recovered = new WALMetaData();
+ try (WALReader reader = new WALReader(logFile, true)) {
+ long previousOffset = 0;
+ while (reader.hasNext()) {
+ WALEntry entry = reader.next();
+ long offset = reader.getLogicalReadOffset();
+ long searchIndex =
+ entry.getType().needSearch() && entry.getValue() instanceof
SearchNode searchNode
+ ? searchNode.getSearchIndex()
+ : ConsensusReqReader.DEFAULT_SEARCH_INDEX;
+ // Use bytes consumed, since reserializing a legacy entry may change
its encoded size.
+ recovered.add(
+ Math.toIntExact(offset - previousOffset), searchIndex,
entry.getMemTableId());
+ recovered.setTruncateOffSet(reader.getWALCurrentReadOffset());
+ previousOffset = offset;
+ }
+ if (!recovered.getBuffersSize().isEmpty() ||
!reader.isFileCorrupted()) {
+ recovered.recoveredFromEntries = true;
+ return recovered;
+ }
+ }
+ // Do not turn an unreadable nonempty file into an empty metadata
result, which would let
+ // WAL cleanup delete it. The suffix also excludes it from subsequent
WAL enumeration.
+ new WALRepairWriter(logFile).quarantine();
+ throw metadataFailure;
+ }
+ }
+
+ public boolean isRecoveredFromEntries() {
+ return recoveredFromEntries;
+ }
+
+ /** Reads the footer only, so repair can validate it without recursively
triggering recovery. */
+ public static WALMetaData readFromWALFileWithoutRecovery(File logFile,
FileChannel channel)
+ throws IOException {
+ if (WALFileVersion.isEmptyOrHeaderOnly(channel)) {
+ return new WALMetaData();
+ }
+ WALFileVersion version = WALFileVersion.getVersion(channel);
+ if (channel.size() < version.getVersionBytes().length + Integer.BYTES
+ || !isValidMagicString(channel, version)) {
throw new BrokenWALFileException(logFile);
}
@@ -362,12 +431,15 @@ public class WALMetaData implements SerializedSize {
long position;
try {
ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
- WALFileVersion version = WALFileVersion.getVersion(channel);
position = channel.size() - Integer.BYTES -
(version.getVersionBytes().length);
IOUtils.readFully(channel, metadataSizeBuf, position);
metadataSizeBuf.flip();
// load metadata
int metadataSize = metadataSizeBuf.getInt();
+ long dataStart = version == WALFileVersion.V1 ? 0 :
version.getVersionBytes().length;
+ if (metadataSize < FIXED_SERIALIZED_SIZE || metadataSize > position -
dataStart) {
+ throw new BrokenWALFileException(logFile);
+ }
ByteBuffer metadataBuf = ByteBuffer.allocate(metadataSize);
IOUtils.readFully(channel, metadataBuf, position - metadataSize);
metadataBuf.flip();
@@ -394,22 +466,18 @@ public class WALMetaData implements SerializedSize {
return metaData;
}
- private static boolean isValidMagicString(FileChannel channel) throws
IOException {
- // V3 magic string is the longest; read enough bytes to check all versions
- int maxMagicLen =
- Math.max(
- WALFileVersion.V3.getVersionBytes().length,
WALFileVersion.V2.getVersionBytes().length);
- if (channel.size() < maxMagicLen) {
+ private static boolean isValidMagicString(FileChannel channel,
WALFileVersion version)
+ throws IOException {
+ int magicLength = version.getVersionBytes().length;
+ if (channel.size() < magicLength) {
return false;
}
- ByteBuffer magicStringBytes = ByteBuffer.allocate(maxMagicLen);
- IOUtils.readFully(channel, magicStringBytes, channel.size() - maxMagicLen);
+ ByteBuffer magicStringBytes = ByteBuffer.allocate(magicLength);
+ IOUtils.readFully(channel, magicStringBytes, channel.size() - magicLength);
magicStringBytes.flip();
String magicString = new String(magicStringBytes.array(),
StandardCharsets.UTF_8);
- return magicString.contains(WALFileVersion.V3.getVersionString())
- || magicString.contains(WALFileVersion.V2.getVersionString())
- || magicString.contains(WALFileVersion.V1.getVersionString());
+ return version.getVersionString().equals(magicString);
}
public void setTruncateOffSet(long offset) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java
index 0a2c994eb54..c54f3e45bc4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java
@@ -50,6 +50,7 @@ public class WALReader implements Closeable {
private final DataInputStream logStream;
private WALEntry nextEntry;
private boolean fileCorrupted = false;
+ private boolean endMarkerReached = false;
public WALReader(File logFile) throws IOException {
this(logFile, false);
@@ -58,8 +59,9 @@ public class WALReader implements Closeable {
public WALReader(File logFile, boolean fileMayCorrupt) throws IOException {
this.logFile = logFile;
this.fileMayCorrupt = fileMayCorrupt;
- this.walInputStream = new WALInputStream(logFile);
+ this.walInputStream = new WALInputStream(logFile, fileMayCorrupt);
this.logStream = new DataInputStream(walInputStream);
+ this.endMarkerReached = walInputStream.available() == 0;
}
/** Like {@link Iterator#hasNext()}. */
@@ -68,7 +70,7 @@ public class WALReader implements Closeable {
return true;
}
// read WALEntries from log stream
- if (fileCorrupted) {
+ if (fileCorrupted || endMarkerReached) {
return false;
}
try {
@@ -79,6 +81,7 @@ public class WALReader implements Closeable {
}
nextEntry = WALEntry.deserialize(logStream);
if (nextEntry.getType() == WALEntryType.WAL_FILE_INFO_END_MARKER) {
+ endMarkerReached = true;
nextEntry = null;
return false;
}
@@ -103,10 +106,15 @@ public class WALReader implements Closeable {
return walInputStream.getFileCurrentPos();
}
+ /** Returns whether reading stopped because the WAL contents were malformed
or truncated. */
public boolean isFileCorrupted() {
return fileCorrupted;
}
+ public long getLogicalReadOffset() {
+ return walInputStream.getLogicalReadOffset();
+ }
+
/**
* Like {@link Iterator#next()}.
*
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java
index 37e32d85beb..d340b7ce0df 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java
@@ -67,7 +67,9 @@ public class WALWriter extends LogWriter {
}
private synchronized void endFile() throws IOException {
- if (logFile.length() == version.getVersionBytes().length) {
+ // The writer channel is write-only; its known version and size identify
an empty header.
+ if (version != WALFileVersion.V1 && logChannel.size() ==
version.getVersionBytes().length) {
+ // A WAL with no entries is valid and has no marker or metadata trailer
to append.
super.close();
return;
}
@@ -98,7 +100,18 @@ public class WALWriter extends LogWriter {
@Override
public void close() throws IOException {
- endFile();
+ try {
+ endFile();
+ } catch (IOException | RuntimeException e) {
+ // In particular, failed recovery-file sealing must release the handle
before its temporary
+ // file can be removed on Windows, while preserving the original failure.
+ try {
+ super.close();
+ } catch (IOException closeException) {
+ e.addSuppressed(closeException);
+ }
+ throw e;
+ }
super.close();
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
index 0e66509d350..95912857280 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
@@ -656,9 +656,8 @@ public class WALNode implements IWALNode {
public boolean isContainsActiveOrPinnedMemTable(Long versionId) {
Set<Long> memTableIdsOfCurrentWal = buffer.getMemTableIds(versionId);
- // If this set is empty, there is a case where WalEntry has been logged
but not persisted,
- // because WalEntry is persisted asynchronously. In this case, the file
cannot be deleted
- // directly, so it is considered active
+ // A null result means that the WAL is still being written or its
metadata could not be read.
+ // Keep the file in either case because its memTable ids are unknown.
if (memTableIdsOfCurrentWal == null) {
return true;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java
index 30d628dd45e..7efce703656 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java
@@ -130,12 +130,12 @@ public class WALNodeRecoverTask implements Runnable {
.STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_ADD_THIS_FA6ADE22,
logDirectory);
} else {
- // delete this wal node folder
- FileUtils.deleteFileOrDirectory(logDirectory);
- logger.info(
- StorageEngineMessages
-
.STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_SO_DELETE_A17892D9,
- logDirectory);
+ if (cleanupRecoveredDirectory(logDirectory)) {
+ logger.info(
+ StorageEngineMessages
+
.STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_SO_DELETE_A17892D9,
+ logDirectory);
+ }
}
// IoTConsensusV2 will not only delete WAL node folder, but also
register WAL node.
@@ -170,6 +170,7 @@ public class WALNodeRecoverTask implements Runnable {
WALMetaData metaData = new WALMetaData(lastSearchIndex, new ArrayList<>(),
new HashSet<>());
WALFileStatus fileStatus = WALFileStatus.CONTAINS_NONE_SEARCH_INDEX;
try (WALReader walReader = new WALReader(lastWALFile, true)) {
+ long previousLogicalOffset = 0;
while (walReader.hasNext()) {
WALEntry walEntry = walReader.next();
long searchIndex = DEFAULT_SEARCH_INDEX;
@@ -182,13 +183,21 @@ public class WALNodeRecoverTask implements Runnable {
}
}
metaData.setTruncateOffSet(walReader.getWALCurrentReadOffset());
- metaData.add(walEntry.serializedSize(), searchIndex,
walEntry.getMemTableId());
+ long logicalOffset = walReader.getLogicalReadOffset();
+ // Legacy entries may serialize differently in this version; retain
their on-disk sizes.
+ metaData.add(
+ Math.toIntExact(logicalOffset - previousLogicalOffset),
+ searchIndex,
+ walEntry.getMemTableId());
+ previousLogicalOffset = logicalOffset;
}
} catch (Exception e) {
logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_LOGS_SKIP,
lastWALFile, e);
}
// make sure last wal file is correct
- repairWalFileIfBroken(lastWALFile, metaData);
+ if (!repairWalFileIfBroken(lastWALFile, metaData)) {
+ return new long[] {lastVersionId, lastSearchIndex};
+ }
// rename last wal file when file status are inconsistent
if (WALFileUtils.parseStatusCode(lastWALFile.getName()) != fileStatus) {
String targetName =
@@ -203,13 +212,38 @@ public class WALNodeRecoverTask implements Runnable {
return new long[] {lastVersionId, lastSearchIndex};
}
- private static void repairWalFileIfBroken(File walFile, WALMetaData
metaData) {
+ /** Clears recovered logs but retains quarantined files for diagnosis across
restarts. */
+ static boolean cleanupRecoveredDirectory(File directory) {
+ File[] quarantined =
+ directory.listFiles((dir, name) ->
name.matches(".*\\.wal\\.broken(?:\\.\\d+)?"));
+ if (quarantined == null) {
+ return false;
+ }
+ if (quarantined.length == 0) {
+ FileUtils.deleteFileOrDirectory(directory);
+ return true;
+ }
+ File[] logs =
+ directory.listFiles(
+ (dir, name) ->
+ WALFileUtils.walFilenameFilter(dir, name)
+ || CheckpointFileUtils.checkpointFilenameFilter(dir,
name));
+ if (logs != null) {
+ for (File log : logs) {
+ FileUtils.deleteFileOrDirectory(log);
+ }
+ }
+ return false;
+ }
+
+ private static boolean repairWalFileIfBroken(File walFile, WALMetaData
metaData) {
WALRepairWriter walRepairWriter = new WALRepairWriter(walFile);
try {
- walRepairWriter.repair(metaData);
+ return walRepairWriter.repair(metaData);
} catch (IOException e) {
logger.error(StorageEngineMessages.FAIL_TO_RECOVER_WAL_METADATA,
walFile, e);
DataNodeExceptionMetrics.getInstance().recordSuspiciousDiskException(e);
+ return false;
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java
index 46598561a5b..94129d9e220 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java
@@ -19,7 +19,8 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.recover;
-import org.apache.iotdb.commons.utils.IOUtils;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALByteBufReader;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter;
@@ -28,7 +29,11 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
-import java.nio.charset.StandardCharsets;
+import java.nio.file.AtomicMoveNotSupportedException;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
/** Check whether the wal file is broken and repair it. */
@@ -39,36 +44,110 @@ public class WALRepairWriter {
this.logFile = logFile;
}
- public void repair(WALMetaData metaData) throws IOException {
- // locate broken data
- long truncateSize;
+ /**
+ * Repairs a WAL from the readable prefix. Returns {@code false} when the
file has no recoverable
+ * entry and is moved aside with a {@code .broken} suffix.
+ */
+ public boolean repair(WALMetaData metaData) throws IOException {
+ if (isEmptyOrHeaderOnly()) {
+ return true;
+ }
+
WALFileVersion version = WALFileVersion.getVersion(logFile);
- if (version.getVersionString().equals(readTailMagic(version))) { //
complete file
- return;
- } else { // file with broken magic string
- truncateSize = metaData.getTruncateOffSet();
+ if (hasReadableMetadata()) {
+ return true;
}
- // truncate broken data
- try (FileChannel channel = FileChannel.open(logFile.toPath(),
StandardOpenOption.APPEND)) {
- channel.truncate(truncateSize);
+ // The caller has already scanned the readable entries and supplied their
rebuilt metadata.
+ if (metaData.getBuffersSize().isEmpty()) {
+ quarantine();
+ return false;
}
- // flush metadata
- try (WALWriter walWriter = new WALWriter(logFile, version)) {
- walWriter.updateMetaData(metaData);
+ // A channel offset may include a partially read entry in the same
compressed segment. Rebuild
+ // complete entries in a temporary file instead of truncating at a
read-ahead offset. Publish
+ // only after every entry and the new footer have been written and forced
successfully.
+ Path repaired =
+ Files.createTempFile(logFile.toPath().toAbsolutePath().getParent(),
"wal-repair-", ".tmp");
+ try {
+ try (WALByteBufReader reader = new WALByteBufReader(logFile, metaData)) {
+ if (version == WALFileVersion.V1) {
+ // V1 entries are raw bytes, without the segment framing emitted by
modern WALWriter.
+ try (FileChannel output = FileChannel.open(repaired,
StandardOpenOption.WRITE)) {
+ while (reader.hasNext()) {
+ writeFully(output, reader.next());
+ }
+ ByteBuffer footer =
+ ByteBuffer.allocate(
+ 1
+ + metaData.serializedSize(version)
+ + Integer.BYTES
+ + version.getVersionBytes().length);
+ footer.put(WALEntryType.WAL_FILE_INFO_END_MARKER.getCode());
+ metaData.serialize(footer, version);
+
footer.putInt(metaData.serializedSize(version)).put(version.getVersionBytes()).flip();
+ writeFully(output, footer);
+ output.force(true);
+ }
+ } else {
+ try (WALWriter writer = new WALWriter(repaired.toFile(), version)) {
+ while (reader.hasNext()) {
+ ByteBuffer entry = reader.next();
+ entry.position(entry.limit());
+ writer.write(entry, false);
+ }
+ writer.updateMetaData(metaData);
+ }
+ }
+ }
+ try {
+ Files.move(
+ repaired,
+ logFile.toPath(),
+ StandardCopyOption.ATOMIC_MOVE,
+ StandardCopyOption.REPLACE_EXISTING);
+ } catch (AtomicMoveNotSupportedException e) {
+ Files.move(repaired, logFile.toPath(),
StandardCopyOption.REPLACE_EXISTING);
+ }
+ } finally {
+ Files.deleteIfExists(repaired);
}
+ return true;
}
- private String readTailMagic(WALFileVersion version) throws IOException {
- int size = version.getVersionBytes().length;
- if (logFile.length() < size) {
- return null;
+ private static void writeFully(FileChannel output, ByteBuffer buffer) throws
IOException {
+ while (buffer.hasRemaining()) {
+ output.write(buffer);
}
+ }
+
+ private boolean isEmptyOrHeaderOnly() throws IOException {
try (FileChannel channel = FileChannel.open(logFile.toPath(),
StandardOpenOption.READ)) {
- ByteBuffer magicStringBytes = ByteBuffer.allocate(size);
- IOUtils.readFully(channel, magicStringBytes, channel.size() - size);
- magicStringBytes.flip();
- return new String(magicStringBytes.array(), StandardCharsets.UTF_8);
+ return WALFileVersion.isEmptyOrHeaderOnly(channel);
+ }
+ }
+
+ private boolean hasReadableMetadata() {
+ try (FileChannel channel = FileChannel.open(logFile.toPath(),
StandardOpenOption.READ)) {
+ WALMetaData.readFromWALFileWithoutRecovery(logFile, channel);
+ return true;
+ } catch (IOException | RuntimeException e) {
+ return false;
+ }
+ }
+
+ /** Moves an unrecoverable WAL aside without replacing an earlier
quarantined file. */
+ public void quarantine() throws IOException {
+ int suffix = 0;
+ while (true) {
+ File target = new File(logFile.getPath() + ".broken" + (suffix == 0 ? ""
: "." + suffix));
+ try {
+ // ATOMIC_MOVE may overwrite an existing target on some providers. A
no-replace move
+ // preserves evidence even when another reader chooses the same
quarantine name.
+ Files.move(logFile.toPath(), target.toPath());
+ return;
+ } catch (FileAlreadyExistsException e) {
+ suffix++;
+ }
}
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
index 766e9cb9046..ba34aef1a9e 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
@@ -31,6 +32,7 @@ import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalIn
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRepairWriter;
import
org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
@@ -38,6 +40,7 @@ import org.apache.iotdb.db.utils.constant.TestConstant;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.write.schema.MeasurementSchema;
@@ -47,12 +50,14 @@ import org.junit.Test;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
+import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -86,6 +91,8 @@ public class WALFileTest {
if (walFile.exists()) {
Files.delete(walFile.toPath());
}
+ Files.deleteIfExists(new File(walFile + ".broken").toPath());
+ Files.deleteIfExists(new File(walFile + ".broken.1").toPath());
}
/** Unexpected channel closure must propagate to the buffer instead of
acknowledging a write. */
@@ -250,27 +257,224 @@ public class WALFileTest {
}
@Test
- public void testReadMetadataFromBrokenFile() throws IOException {
- ILogWriter walWriter = new WALWriter(walFile);
- final FileChannel fileChannel1 = FileChannel.open(walFile.toPath());
- assertThrows(IOException.class, () -> WALMetaData.readFromWALFile(walFile,
fileChannel1));
- walWriter.close();
-
- if (!walFile.exists()) {
- Files.createFile(walFile.toPath());
- Files.write(walFile.toPath(),
ByteBuffer.wrap(WALFileVersion.V2.getVersionBytes()).array());
+ public void testReadMetadataFromEmptyFile() throws IOException {
+ try (WALWriter walWriter = new WALWriter(walFile);
+ FileChannel fileChannel = FileChannel.open(walFile.toPath())) {
+ WALMetaData walMetaData = WALMetaData.readFromWALFile(walFile,
fileChannel);
+ assertTrue(walMetaData.getBuffersSize().isEmpty());
+ assertTrue(walMetaData.getMemTablesId().isEmpty());
}
+ assertEquals(WALFileVersion.V3.getVersionBytes().length, walFile.length());
+ try (WALByteBufReader reader = new WALByteBufReader(walFile)) {
+ assertFalse(reader.hasNext());
+ }
+ }
+
+ @Test
+ public void testReadMetadataFromTruncatedFile() throws IOException {
+ Files.write(walFile.toPath(), WALFileVersion.V3.getVersionBytes());
+ Files.write(walFile.toPath(), new byte[] {1}, StandardOpenOption.APPEND);
+
+ try (FileChannel fileChannel = FileChannel.open(walFile.toPath())) {
+ assertThrows(IOException.class, () ->
WALMetaData.readFromWALFile(walFile, fileChannel));
+ }
+ assertFalse(walFile.exists());
+ assertTrue(new File(walFile + ".broken").exists());
+ }
+
+ @Test
+ public void testRecoverEntriesWithCorruptedMetadataLength() throws Exception
{
+ for (WALFileVersion version : new WALFileVersion[] {WALFileVersion.V2,
WALFileVersion.V3}) {
+ for (int length : new int[] {-1, Integer.MAX_VALUE, 0, 12}) {
+ Files.deleteIfExists(walFile.toPath());
+ WALEntry entry = new WALInfoEntry(42, getInsertRowNode(devicePath));
+ WALMetaData metadata = new WALMetaData();
+ metadata.add(entry.serializedSize(), -1, 42);
+ WALByteBufferForTest buffer =
+ new
WALByteBufferForTest(ByteBuffer.allocate(entry.serializedSize()));
+ entry.serialize(buffer);
+ try (WALWriter writer = new WALWriter(walFile, version)) {
+ writer.write(buffer.getBuffer(), metadata);
+ }
+ byte[] bytes = Files.readAllBytes(walFile.toPath());
+ ByteBuffer.wrap(bytes).putInt(bytes.length -
version.getVersionBytes().length - 4, length);
+ Files.write(walFile.toPath(), bytes);
+ try (WALByteBufReader reader = new WALByteBufReader(walFile)) {
+ assertTrue(reader.getMetaData().isRecoveredFromEntries());
+ assertEquals(Collections.singleton(42L),
reader.getMetaData().getMemTablesId());
+ assertTrue(reader.hasNext());
+ assertEquals(
+ entry,
+ WALEntry.deserialize(
+ new DataInputStream(new
ByteArrayInputStream(reader.next().array()))));
+ assertFalse(reader.hasNext());
+ }
+ // Readers reconstruct an in-memory prefix without replacing bytes
under other readers.
+ assertArrayEquals(bytes, Files.readAllBytes(walFile.toPath()));
+ }
+ }
+ }
+
+ @Test
+ public void testRecoveryRejectsOversizedCompressedSegment() throws Exception
{
+ // Include the declared payload so rejection is caused by its logical
size, not a short read.
+ ByteBuffer bytes =
+ ByteBuffer.allocate(WALFileVersion.V3.getVersionBytes().length + 2 + 2
* Integer.BYTES);
+ bytes.put(WALFileVersion.V3.getVersionBytes());
+ bytes.put(CompressionType.LZ4.serialize());
+ bytes.putInt(1);
+ bytes.putInt(64 * 1024 * 1024);
+ bytes.put((byte) 0);
+ Files.write(walFile.toPath(), bytes.array());
+
+ try (WALInputStream input = new WALInputStream(walFile, true)) {
+ // A decompressor failure is wrapped in IOException; this must fail
before reaching it.
+ assertThrows(EOFException.class, input::read);
+ }
+ }
+
+ @Test
+ public void testUnrecoverableFileDoesNotOverwriteQuarantine() throws
Exception {
+ byte[] previous = new byte[] {9, 8, 7};
+ Files.write(new File(walFile + ".broken").toPath(), previous);
+ byte[] corrupt = new byte[] {99, 98};
+ Files.write(walFile.toPath(), corrupt);
+ assertThrows(IOException.class, () -> new WALByteBufReader(walFile));
+ assertArrayEquals(previous, Files.readAllBytes(new File(walFile +
".broken").toPath()));
+ assertArrayEquals(corrupt, Files.readAllBytes(new File(walFile +
".broken.1").toPath()));
+ assertFalse(walFile.exists());
+ }
+
+ @Test
+ public void testRecoverPrefixFromTruncatedSegment() throws Exception {
+ WALEntry entry = new WALInfoEntry(42, getInsertRowNode(devicePath));
+ WALByteBufferForTest buffer =
+ new WALByteBufferForTest(ByteBuffer.allocate(entry.serializedSize() *
2));
+ entry.serialize(buffer);
+ entry.serialize(buffer);
+ long dataEnd;
+ try (WALWriter writer = new WALWriter(walFile)) {
+ writer.write(buffer.getBuffer(), false);
+ dataEnd = writer.getOffset();
+ }
+ // Preserve the declared segment size but cut the second entry in half,
losing the footer too.
+ try (FileChannel channel = FileChannel.open(walFile.toPath(),
StandardOpenOption.WRITE)) {
+ channel.truncate(dataEnd - entry.serializedSize() / 2);
+ }
+ byte[] original = Files.readAllBytes(walFile.toPath());
+ WALMetaData recovered;
+ try (WALByteBufReader reader = new WALByteBufReader(walFile)) {
+ recovered = reader.getMetaData();
+ assertEquals(
+ Collections.singletonList(entry.serializedSize()),
reader.getMetaData().getBuffersSize());
+ assertEquals(
+ entry,
+ WALEntry.deserialize(
+ new DataInputStream(new
ByteArrayInputStream(reader.next().array()))));
+ assertFalse(reader.hasNext());
+ }
+ assertArrayEquals(original, Files.readAllBytes(walFile.toPath()));
+ assertTrue(new WALRepairWriter(walFile).repair(recovered));
+ try (FileChannel channel = FileChannel.open(walFile.toPath())) {
+ assertEquals(
+ Collections.singletonList(entry.serializedSize()),
+ WALMetaData.readFromWALFileWithoutRecovery(walFile,
channel).getBuffersSize());
+ }
+ try (WALReader reader = new WALReader(walFile)) {
+ assertEquals(entry, reader.next());
+ assertFalse(reader.hasNext());
+ assertFalse(reader.isFileCorrupted());
+ }
+ }
+
+ @Test
+ public void testRecoverLegacyEntries() throws Exception {
+ WALEntry entry = new WALInfoEntry(42, getInsertRowNode(devicePath));
+ WALByteBufferForTest buffer =
+ new WALByteBufferForTest(ByteBuffer.allocate(entry.serializedSize() +
1));
+ entry.serialize(buffer);
+ buffer.put(WALEntryType.DELETE_DATA_NODE.getCode());
+ // V1 uses raw entry bytes; the final entry is deliberately incomplete and
has no footer.
+ Files.write(walFile.toPath(), buffer.getBuffer().array());
+ WALMetaData recovered;
+ try (WALByteBufReader reader = new WALByteBufReader(walFile)) {
+ recovered = reader.getMetaData();
+ assertEquals(Collections.singleton(42L), recovered.getMemTablesId());
+ assertEquals(entry.serializedSize(), reader.next().remaining());
+ assertFalse(reader.hasNext());
+ }
+ assertTrue(new WALRepairWriter(walFile).repair(recovered));
+ assertEquals(WALFileVersion.V1, WALFileVersion.getVersion(walFile));
+ try (WALReader reader = new WALReader(walFile)) {
+ assertEquals(entry, reader.next());
+ assertFalse(reader.hasNext());
+ assertFalse(reader.isFileCorrupted());
+ }
+ }
+
+ @Test
+ public void testRecoverCompressedEntries() throws Exception {
+ CompressionType originalCompression =
+ IoTDBDescriptor.getInstance().getConfig().getWALCompressionAlgorithm();
try {
- FileChannel fileChannel2 = FileChannel.open(walFile.toPath());
- WALMetaData walMetaData = WALMetaData.readFromWALFile(walFile,
fileChannel2);
- fileChannel2.close();
- } catch (Exception e) {
+
IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(CompressionType.LZ4);
+ WALEntry entry = new WALInfoEntry(42, getInsertRowNode(devicePath));
+ int count = 400;
+ WALByteBufferForTest buffer =
+ new WALByteBufferForTest(ByteBuffer.allocate(entry.serializedSize()
* count));
+ WALMetaData metadata = new WALMetaData();
+ for (int i = 0; i < count; i++) {
+ entry.serialize(buffer);
+ metadata.add(entry.serializedSize(), -1, 42);
+ }
+ try (WALWriter writer = new WALWriter(walFile)) {
+
writer.setCompressedByteBuffer(ByteBuffer.allocate(buffer.getBuffer().capacity()
* 2));
+ writer.write(buffer.getBuffer(), metadata);
+ }
+ byte[] bytes = Files.readAllBytes(walFile.toPath());
assertEquals(
- "Broken wal file "
- + walFile.getPath()
- + ", size "
- + WALFileVersion.V2.getVersionBytes().length,
- e.getMessage());
+ CompressionType.LZ4.serialize(),
bytes[WALFileVersion.V3.getVersionBytes().length]);
+ Files.write(walFile.toPath(), Arrays.copyOf(bytes, bytes.length - 1));
+ WALMetaData recovered;
+ try (WALByteBufReader reader = new WALByteBufReader(walFile)) {
+ recovered = reader.getMetaData();
+ assertEquals(count, recovered.getBuffersSize().size());
+ for (int i = 0; i < count; i++) {
+ assertEquals(
+ entry,
+ WALEntry.deserialize(
+ new DataInputStream(new
ByteArrayInputStream(reader.next().array()))));
+ }
+ assertFalse(reader.hasNext());
+ }
+ assertTrue(new WALRepairWriter(walFile).repair(recovered));
+ try (WALReader reader = new WALReader(walFile)) {
+ for (int i = 0; i < count; i++) {
+ assertEquals(entry, reader.next());
+ }
+ assertFalse(reader.hasNext());
+ assertFalse(reader.isFileCorrupted());
+ }
+ } finally {
+
IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(originalCompression);
+ }
+ }
+
+ @Test
+ public void testEmptyFilesAcrossVersions() throws Exception {
+ for (byte[] bytes :
+ new byte[][] {
+ new byte[0], WALFileVersion.V2.getVersionBytes(),
WALFileVersion.V3.getVersionBytes()
+ }) {
+ Files.write(walFile.toPath(), bytes);
+ try (WALReader reader = new WALReader(walFile, true)) {
+ assertFalse(reader.hasNext());
+ assertFalse(reader.isFileCorrupted());
+ }
+ try (WALByteBufReader reader = new WALByteBufReader(walFile)) {
+ assertFalse(reader.hasNext());
+ }
+ assertArrayEquals(bytes, Files.readAllBytes(walFile.toPath()));
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriterTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriterTest.java
index 315e963fbeb..ea0a615d984 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriterTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriterTest.java
@@ -40,6 +40,7 @@ import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.After;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Test;
import java.io.File;
@@ -48,9 +49,11 @@ import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
+import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.HashSet;
+import java.util.stream.Stream;
public class WALRepairWriterTest {
private final File logFile =
@@ -58,9 +61,19 @@ public class WALRepairWriterTest {
TestConstant.BASE_OUTPUT_PATH.concat(
WALFileUtils.getLogFileName(1, 1,
WALFileStatus.CONTAINS_SEARCH_INDEX)));
+ @Before
+ public void setUp() throws IOException {
+ Files.createDirectories(logFile.toPath().getParent());
+ }
+
@After
public void tearDown() throws Exception {
logFile.delete();
+ File brokenFile = new File(logFile.getPath() + ".broken");
+ brokenFile.delete();
+ for (int suffix = 1; suffix < 10; suffix++) {
+ new File(logFile.getPath() + ".broken." + suffix).delete();
+ }
}
@Test
@@ -71,19 +84,10 @@ public class WALRepairWriterTest {
WALMetaData walMetaData = new WALMetaData(firstSearchIndex, new
ArrayList<>(), new HashSet<>());
// repair
new WALRepairWriter(logFile).repair(walMetaData);
- // verify file, marker(header size + marker buffer size) + metadata(search
index + size number)
- // + metadata size + head magic
- // string + tail magic string
- // empty file will be assumed as V1 (because of no header magic)
- Assert.assertEquals(
- (Byte.BYTES + Integer.BYTES + Byte.BYTES)
- + (Long.BYTES + Integer.BYTES)
- + Integer.BYTES
- + WALFileVersion.V1.getVersionBytes().length,
- logFile.length());
+ Assert.assertEquals(0, logFile.length());
try (WALByteBufReader reader = new WALByteBufReader(logFile)) {
Assert.assertFalse(reader.hasNext());
- Assert.assertEquals(firstSearchIndex, reader.getFirstSearchIndex());
+ Assert.assertTrue(reader.getMetaData().getMemTablesId().isEmpty());
}
}
@@ -97,20 +101,10 @@ public class WALRepairWriterTest {
long firstSearchIndex =
WALFileUtils.parseStartSearchIndex(logFile.getName());
WALMetaData walMetaData = new WALMetaData(firstSearchIndex, new
ArrayList<>(), new HashSet<>());
// repair
- new WALRepairWriter(logFile).repair(walMetaData);
- // verify file, marker(header size + marker buffer size) + metadata(search
index + size number)
- // + metadata size + magic string
- // file too small will be assumed as V1 (because of no header magic)
- Assert.assertEquals(
- (Byte.BYTES + Integer.BYTES + Byte.BYTES)
- + (Long.BYTES + Integer.BYTES)
- + Integer.BYTES
- + WALFileVersion.V1.getVersionBytes().length,
- logFile.length());
- try (WALByteBufReader reader = new WALByteBufReader(logFile)) {
- Assert.assertFalse(reader.hasNext());
- Assert.assertEquals(firstSearchIndex, reader.getFirstSearchIndex());
- }
+ Assert.assertFalse(new WALRepairWriter(logFile).repair(walMetaData));
+ Assert.assertFalse(logFile.exists());
+ Assert.assertArrayEquals(
+ new byte[] {1}, Files.readAllBytes(new File(logFile +
".broken").toPath()));
}
@Test
@@ -187,6 +181,102 @@ public class WALRepairWriterTest {
}
}
+ @Test
+ public void testUnrecoverableFileIsQuarantined() throws IOException {
+ Files.write(logFile.toPath(), new byte[] {1, 2, 3, 4});
+
+ Assert.assertFalse(new WALRepairWriter(logFile).repair(new WALMetaData()));
+ Assert.assertFalse(logFile.exists());
+ Assert.assertTrue(new File(logFile.getPath() + ".broken").exists());
+ }
+
+ @Test
+ public void testCorruptedMetadataIsRebuilt() throws IOException,
IllegalPathException {
+ WALMetaData walMetaData = new WALMetaData();
+ WALEntry walEntry = new WALInfoEntry(1, getInsertRowNode());
+ int size = walEntry.serializedSize();
+ WALByteBufferForTest buffer = new
WALByteBufferForTest(ByteBuffer.allocate(size));
+ walEntry.serialize(buffer);
+ walMetaData.add(size, 1, walEntry.getMemTableId());
+
+ long truncateOffset;
+ try (WALWriter walWriter = new WALWriter(logFile)) {
+ walWriter.write(buffer.getBuffer(), walMetaData);
+ truncateOffset = walWriter.getOffset();
+ }
+
+ byte[] fileBytes = Files.readAllBytes(logFile.toPath());
+ int metadataSizeOffset =
+ fileBytes.length - WALFileVersion.V3.getVersionBytes().length -
Integer.BYTES;
+ int metadataSize = ByteBuffer.wrap(fileBytes, metadataSizeOffset,
Integer.BYTES).getInt();
+ int metadataOffset = metadataSizeOffset - metadataSize;
+ ByteBuffer.wrap(fileBytes).putInt(metadataOffset + Long.BYTES, -1);
+ Files.write(logFile.toPath(), fileBytes);
+
+ WALMetaData recoveredMetadata = walMetaData.copy();
+ recoveredMetadata.setTruncateOffSet(truncateOffset);
+ Assert.assertTrue(new WALRepairWriter(logFile).repair(recoveredMetadata));
+
+ try (WALByteBufReader reader = new WALByteBufReader(logFile)) {
+ Assert.assertTrue(reader.hasNext());
+ Assert.assertEquals(size, reader.next().capacity());
+ Assert.assertFalse(reader.hasNext());
+ }
+ }
+
+ @Test
+ public void testFailedRepairPreservesOriginalFile() throws Exception {
+ WALEntry entry = new WALInfoEntry(1, getInsertRowNode());
+ WALByteBufferForTest buffer =
+ new WALByteBufferForTest(ByteBuffer.allocate(entry.serializedSize()));
+ entry.serialize(buffer);
+ long dataEnd;
+ try (WALWriter writer = new WALWriter(logFile)) {
+ writer.write(buffer.getBuffer(), false);
+ dataEnd = writer.getOffset();
+ }
+ try (FileChannel channel = FileChannel.open(logFile.toPath(),
StandardOpenOption.WRITE)) {
+ channel.truncate(dataEnd);
+ }
+ byte[] original = Files.readAllBytes(logFile.toPath());
+ WALMetaData invalidSnapshot = new WALMetaData();
+ invalidSnapshot.add(entry.serializedSize(), 1, 1);
+ invalidSnapshot.add(entry.serializedSize(), 2, 1);
+ // A stale snapshot requests one entry beyond EOF. The original must
survive a failed rewrite.
+ Assert.assertThrows(
+ IOException.class, () -> new
WALRepairWriter(logFile).repair(invalidSnapshot));
+ Assert.assertArrayEquals(original, Files.readAllBytes(logFile.toPath()));
+ try (Stream<Path> files = Files.list(logFile.toPath().getParent())) {
+ Assert.assertFalse(
+ files.anyMatch(path ->
path.getFileName().toString().startsWith("wal-repair-")));
+ }
+ }
+
+ @Test
+ public void testStartupCleanupRetainsQuarantinedFile() throws Exception {
+ Path directory = Files.createTempDirectory(logFile.toPath().getParent(),
"wal-cleanup-");
+ Path broken = directory.resolve(logFile.getName() + ".broken.1");
+ Path wal = directory.resolve(logFile.getName());
+ Path checkpoint = directory.resolve("_0.checkpoint");
+ try {
+ Files.write(broken, new byte[] {1, 2});
+ Files.write(wal, new byte[] {3});
+ Files.write(checkpoint, new byte[] {4});
+
Assert.assertFalse(WALNodeRecoverTask.cleanupRecoveredDirectory(directory.toFile()));
+ Assert.assertArrayEquals(new byte[] {1, 2}, Files.readAllBytes(broken));
+ Assert.assertFalse(Files.exists(wal));
+ Assert.assertFalse(Files.exists(checkpoint));
+ Files.delete(broken);
+
Assert.assertTrue(WALNodeRecoverTask.cleanupRecoveredDirectory(directory.toFile()));
+ Assert.assertFalse(Files.exists(directory));
+ } finally {
+ Files.deleteIfExists(broken);
+ Files.deleteIfExists(wal);
+ Files.deleteIfExists(checkpoint);
+ Files.deleteIfExists(directory);
+ }
+ }
+
public static InsertRowNode getInsertRowNode() throws IllegalPathException {
String devicePath = "root.test_sg.test_d";
long time = 110L;
