This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch fix_empty_wal_read in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7f0f8e4372e37b8feeb9d30f783e6d427cd8f9b8 Author: Tian Jiang <[email protected]> AuthorDate: Tue Sep 22 15:28:16 2026 +0800 Recover WAL metadata from readable entries and quarantine unrecoverable files --- .../dataregion/wal/buffer/WALBuffer.java | 11 +- .../dataregion/wal/io/WALByteBufReader.java | 4 +- .../dataregion/wal/io/WALFileVersion.java | 19 ++ .../dataregion/wal/io/WALInputStream.java | 70 ++++++- .../dataregion/wal/io/WALMetaData.java | 98 +++++++-- .../storageengine/dataregion/wal/io/WALReader.java | 16 +- .../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 | 221 +++++++++++++++++++-- .../wal/recover/WALRepairWriterTest.java | 140 ++++++++++--- 12 files changed, 667 insertions(+), 113 deletions(-) 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..9d6a5176291 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 @@ -56,7 +56,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); 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..6dcb9f5b326 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 @@ -59,16 +59,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 +91,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 +139,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 +171,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 +183,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 +191,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 +253,18 @@ 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); + } try { if (segmentInfo.compressionType != CompressionType.UNCOMPRESSED) { // A compressed segment @@ -352,10 +394,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 befcf58c632..f43ddd8f000 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,12 +70,13 @@ public class WALReader implements Closeable { return true; } // read WALEntries from log stream - if (fileCorrupted) { + if (fileCorrupted || endMarkerReached) { return false; } try { nextEntry = WALEntry.deserialize(logStream); if (nextEntry.getType() == WALEntryType.WAL_FILE_INFO_END_MARKER) { + endMarkerReached = true; nextEntry = null; return false; } @@ -98,6 +101,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..561d0605ef9 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; @@ -53,6 +56,7 @@ 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 +90,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 +256,206 @@ 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 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;
