This is an automated email from the ASF dual-hosted git repository.
HTHou 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 9eaf5f7baf8 Add optional WAL and checkpoint format extension
interfaces (#19002)
9eaf5f7baf8 is described below
commit 9eaf5f7baf8eeb5da9f8cb0ce16ee1cef376fbb2
Author: Haonan <[email protected]>
AuthorDate: Mon Oct 26 18:32:30 2026 +0800
Add optional WAL and checkpoint format extension interfaces (#19002)
---
.../iotdb/db/i18n/StorageEngineMessages.java | 6 +
.../iotdb/db/i18n/StorageEngineMessages.java | 6 +
.../iotdb/db/storageengine/StorageEngine.java | 8 +
.../dataregion/wal/io/CheckpointReader.java | 6 +
.../dataregion/wal/io/CheckpointWriter.java | 2 +-
.../storageengine/dataregion/wal/io/LogWriter.java | 38 +-
.../dataregion/wal/io/WALFileContext.java | 56 ++
.../dataregion/wal/io/WALFileVersion.java | 17 +-
.../dataregion/wal/io/WALFormatException.java | 54 ++
.../dataregion/wal/io/WALFormatProvider.java | 51 ++
.../dataregion/wal/io/WALFormats.java | 149 +++++
.../dataregion/wal/io/WALInputStream.java | 97 ++-
.../dataregion/wal/io/WALMetaData.java | 6 +
.../storageengine/dataregion/wal/io/WALReader.java | 9 +
.../storageengine/dataregion/wal/io/WALWriter.java | 19 +-
.../dataregion/wal/recover/WALNodeRecoverTask.java | 130 +++--
.../dataregion/wal/recover/WALRecoverManager.java | 56 +-
.../dataregion/wal/recover/WALRepairWriter.java | 4 +
.../dataregion/wal/utils/WALFileUtils.java | 6 +-
.../dataregion/wal/io/WALFormatExtensionTest.java | 649 +++++++++++++++++++++
.../dataregion/wal/io/WALFormatsTest.java | 150 +++++
21 files changed, 1445 insertions(+), 74 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 25b73c30152..e669b1a4b80 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
@@ -20,6 +20,12 @@
package org.apache.iotdb.db.i18n;
public final class StorageEngineMessages {
+ public static final String
EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4 =
+ "Unsupported or unavailable WAL format";
+ public static final String
EXCEPTION_MULTIPLE_WAL_FORMAT_PROVIDERS_ARE_CONFIGURED_DD78877D =
+ "Multiple WAL format providers are configured";
+ public static final String EXCEPTION_INVALID_WAL_LOGICAL_POSITION_8CF34AFE =
+ "Invalid WAL logical position";
public static final String
LOG_REBUILT_UNREADABLE_WAL_FOOTER_FILE_ARG_VERSION_ARG_ORIGINALFILEBYTES_ARG_WRITERPROGRESS_ARG_1DBF836E
=
"Rebuilt unreadable WAL footer: file={}, version={},
originalFileBytes={}, writerProgress={}.";
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 93f9b29b6f8..5f11f94560d 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
@@ -20,6 +20,12 @@
package org.apache.iotdb.db.i18n;
public final class StorageEngineMessages {
+ public static final String
EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4 =
+ "WAL 格式不支持或对应实现不可用";
+ public static final String
EXCEPTION_MULTIPLE_WAL_FORMAT_PROVIDERS_ARE_CONFIGURED_DD78877D =
+ "配置了多个 WAL 格式 Provider";
+ public static final String EXCEPTION_INVALID_WAL_LOGICAL_POSITION_8CF34AFE =
+ "WAL 逻辑位置无效";
public static final String
LOG_REBUILT_UNREADABLE_WAL_FOOTER_FILE_ARG_VERSION_ARG_ORIGINALFILEBYTES_ARG_WRITERPROGRESS_ARG_1DBF836E
=
"已重建不可读的 WAL
footer：file={}，version={}，originalFileBytes={}，writerProgress={}。";
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index f66a76c9f95..36bb9679f81 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -77,6 +77,7 @@ import
org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy;
import
org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy.DirectFlushPolicy;
import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager;
import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALException;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormatException;
import
org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager;
import org.apache.iotdb.db.storageengine.load.LoadTsFilePieceNodeAssembler;
import org.apache.iotdb.db.storageengine.load.active.ActiveLoadAgent;
@@ -231,6 +232,13 @@ public class StorageEngine implements IService {
WALRecoverManager.getInstance().recover();
} catch (WALException e) {
LOGGER.error(StorageEngineMessages.FAIL_TO_RECOVER_WAL, e);
+ if (WALFormatException.isFatal(e)) {
+ cachedThreadPool.shutdownNow();
+ StartupException failure =
+ new StartupException(StorageEngineMessages.FAIL_TO_RECOVER_WAL);
+ failure.initCause(e);
+ throw failure;
+ }
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointReader.java
index 7069781fd8b..dfd13cfd571 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointReader.java
@@ -29,6 +29,7 @@ import java.io.DataInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.List;
@@ -53,6 +54,8 @@ public class CheckpointReader {
Checkpoint checkpoint = Checkpoint.deserialize(logStream);
checkpoints.add(checkpoint);
}
+ } catch (WALFormatException e) {
+ throw new UncheckedIOException(e);
} catch (EOFException e) {
logger.debug(
StorageEngineMessages
@@ -60,6 +63,9 @@ public class CheckpointReader {
logFile,
e.getMessage());
} catch (IOException e) {
+ if (WALFormatException.isFatal(e)) {
+ throw new UncheckedIOException(e);
+ }
logger.warn(
StorageEngineMessages
.STORAGE_LOG_MEET_ERROR_WHEN_READING_CHECKPOINT_FILE_SKIP_BROKEN_CHECKPOINTS_DADF9E9D,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointWriter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointWriter.java
index f31f2e9b846..1f8cd943360 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointWriter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointWriter.java
@@ -27,6 +27,6 @@ import java.io.IOException;
/** CheckpointWriter writes the binary {@link Checkpoint} into .checkpoint
file. */
public class CheckpointWriter extends LogWriter {
public CheckpointWriter(File logFile) throws IOException {
- super(logFile, WALFileVersion.V2);
+ super(logFile, WALFormats.writeVersion(WALFileVersion.V2));
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
index 3652c826682..abf96cb19c3 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
@@ -41,6 +41,9 @@ public abstract class LogWriter implements ILogWriter {
protected final File logFile;
protected final FileChannel logChannel;
protected long originalSize = 0;
+ protected WALFileContext fileContext;
+ protected long blockIndex;
+ protected long dataStartOffset;
/**
* 1 byte for whether enable compression, 4 byte for compressedSize, 4 byte
for uncompressedSize
@@ -68,14 +71,17 @@ public abstract class LogWriter implements ILogWriter {
StandardOpenOption.WRITE,
StandardOpenOption.APPEND);
try {
- if (logChannel.size() == 0
+ if (version == WALFileVersion.EXTENDED) {
+ fileContext = WALFormats.create(logFile, logChannel);
+ } else if (logChannel.size() == 0
&& (version == WALFileVersion.V2 || version == WALFileVersion.V3)) {
ByteBuffer magic = ByteBuffer.wrap(version.getVersionBytes());
while (magic.hasRemaining()) {
logChannel.write(magic);
}
}
- } catch (IOException e) {
+ dataStartOffset = logChannel.position();
+ } catch (IOException | RuntimeException | Error e) {
// A full disk can fail initialization after open() succeeds. Release
the orphan channel
// before the owner retries creating the successor.
try {
@@ -83,6 +89,13 @@ public abstract class LogWriter implements ILogWriter {
} catch (IOException closeException) {
e.addSuppressed(closeException);
}
+ if (fileContext != null) {
+ try {
+ fileContext.close();
+ } catch (RuntimeException | Error closeException) {
+ e.addSuppressed(closeException);
+ }
+ }
throw e;
}
}
@@ -132,6 +145,19 @@ public abstract class LogWriter implements ILogWriter {
WritingMetrics.getInstance().recordCompressWALBufferCost(System.nanoTime() -
startTime);
}
startTime = System.nanoTime();
+ if (fileContext != null) {
+ int written =
+ fileContext.writeData(
+ logChannel,
+ blockIndex,
+ compressed ? compressionType : CompressionType.UNCOMPRESSED,
+ uncompressedSize,
+ buffer);
+ blockIndex++;
+ WritingMetrics.getInstance()
+ .recordWroteWALBuffer(uncompressedSize, written, System.nanoTime() -
startTime);
+ return ((double) written / uncompressedSize);
+ }
headerBuffer.flip();
while (headerBuffer.hasRemaining()) {
logChannel.write(headerBuffer);
@@ -183,7 +209,13 @@ public abstract class LogWriter implements ILogWriter {
logChannel.force(true);
}
} finally {
- logChannel.close();
+ try {
+ logChannel.close();
+ } finally {
+ if (fileContext != null) {
+ fileContext.close();
+ }
+ }
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileContext.java
new file mode 100644
index 00000000000..8c925cf9e68
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileContext.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+
+/**
+ * One extension-format file. The caller owns the channel; the context owns
only format resources.
+ * Read methods return decoded segments and must distinguish an incomplete
tail (EOFException) from
+ * an unrecoverable format failure (WALFormatException).
+ */
+public interface WALFileContext extends AutoCloseable {
+ int writeData(
+ FileChannel channel,
+ long index,
+ CompressionType compression,
+ int uncompressedSize,
+ ByteBuffer data)
+ throws IOException;
+
+ /** Returns an uncompressed segment, or throws EOFException at metadata/the
incomplete tail. */
+ ByteBuffer readSegment(FileChannel channel, long endOffset, long index)
throws IOException;
+
+ /** Writes the format-specific metadata frame and trailer after the final
data segment. */
+ void seal(FileChannel channel, WALMetaData metadata, long index) throws
IOException;
+
+ /**
+ * Releases a returned segment, including any sensitive scratch state;
accepts null. The provider
+ * owns its allocation and may recycle it. The caller must not separately
free its native memory.
+ */
+ void release(ByteBuffer data);
+
+ /** Releases format resources; repeated calls must be safe. The channel
belongs to the caller. */
+ @Override
+ void close();
+}
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 f643c78e8dd..c51821b3fb4 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
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.io;
import org.apache.iotdb.commons.utils.IOUtils;
+import org.apache.iotdb.db.i18n.StorageEngineMessages;
import java.io.File;
import java.io.IOException;
@@ -29,19 +30,30 @@ import java.nio.charset.StandardCharsets;
public enum WALFileVersion {
V1("WAL"),
V2("V2-WAL"),
- V3("V3-WAL");
+ V3("V3-WAL"),
+ /** A dispatch marker only; its on-disk magic and layout belong to the
installed provider. */
+ EXTENDED(null);
private final String versionString;
private byte[] versionBytes;
public String getVersionString() {
+ requireBuiltInVersion();
return versionString;
}
public byte[] getVersionBytes() {
+ requireBuiltInVersion();
return versionBytes;
}
+ private void requireBuiltInVersion() {
+ if (this == EXTENDED) {
+ throw new IllegalStateException(
+
StorageEngineMessages.EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4);
+ }
+ }
+
WALFileVersion(String versionString) {
this.versionString = versionString;
if (versionString != null) {
@@ -73,6 +85,9 @@ public enum WALFileVersion {
return version;
}
}
+ if (WALFormats.recognizes(channel)) {
+ return EXTENDED;
+ }
// v1 by default
return V1;
} finally {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatException.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatException.java
new file mode 100644
index 00000000000..100a68e1be4
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatException.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+
+import org.apache.iotdb.db.i18n.StorageEngineMessages;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Set;
+
+/** A format/policy failure that recovery must never downgrade to tail
truncation or skip. */
+public class WALFormatException extends IOException {
+ public WALFormatException(String message) {
+ super(message);
+ }
+
+ public WALFormatException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public static WALFormatException unsupported() {
+ return new WALFormatException(
+
StorageEngineMessages.EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4);
+ }
+
+ public static boolean isFatal(Throwable failure) {
+ Set<Throwable> seen = Collections.newSetFromMap(new IdentityHashMap<>());
+ for (Throwable current = failure;
+ current != null && seen.add(current);
+ current = current.getCause()) {
+ if (current instanceof WALFormatException) {
+ return true;
+ }
+ }
+ return false;
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatProvider.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatProvider.java
new file mode 100644
index 00000000000..17ada89c5ab
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatProvider.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+
+/** Optional WAL/checkpoint format implementation, without prescribing
algorithms or key storage. */
+public interface WALFormatProvider {
+ /**
+ * Stable magic in the reserved {@code V<digits>-WAL} namespace, independent
of the write policy.
+ * It must not reuse a built-in version, so missing providers can be
rejected without guessing V1.
+ */
+ byte[] getMagic();
+
+ boolean isEnabled();
+
+ /** Initializes a new file, including its magic and header; does not take
channel ownership. */
+ WALFileContext create(File file, FileChannel channel) throws IOException;
+
+ /** Opens an existing extension-format file, leaving the channel at its
first segment. */
+ WALFileContext open(File file, FileChannel channel) throws IOException;
+
+ WALMetaData readMetadata(File file, FileChannel channel) throws IOException;
+
+ /** Enforces product policy before opening either a legacy or
extension-format file. */
+ default void validateRead(File file, WALFileVersion version) throws
IOException {}
+
+ /** Checks the file before replay or cleanup may have destructive side
effects. */
+ void validateForRecovery(File file) throws IOException;
+
+ /** Product-specific diagnostics, called once when recovery terminates with
a fatal failure. */
+ default void onRecoveryFailure(Throwable failure) {}
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormats.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormats.java
new file mode 100644
index 00000000000..8691e467096
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormats.java
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+
+import org.apache.iotdb.commons.utils.IOUtils;
+import org.apache.iotdb.db.i18n.StorageEngineMessages;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.ServiceLoader;
+
+/** Dispatches optional formats without exposing their binary layouts to the
standard WAL path. */
+public final class WALFormats {
+ private WALFormats() {}
+
+ private static class Holder {
+ private static final WALFormatProvider PROVIDER =
+ selectProvider(ServiceLoader.load(WALFormatProvider.class).iterator());
+ }
+
+ static WALFormatProvider selectProvider(Iterator<WALFormatProvider>
providers) {
+ if (!providers.hasNext()) {
+ return null;
+ }
+ WALFormatProvider provider = providers.next();
+ if (providers.hasNext()) {
+ throw new IllegalStateException(
+
StorageEngineMessages.EXCEPTION_MULTIPLE_WAL_FORMAT_PROVIDERS_ARE_CONFIGURED_DD78877D);
+ }
+ byte[] magic = provider.getMagic();
+ if (magic == null
+ || magic.length > 64
+ || !new String(magic, StandardCharsets.US_ASCII).matches("V[0-9]+-WAL")
+ || Arrays.equals(magic, WALFileVersion.V2.getVersionBytes())
+ || Arrays.equals(magic, WALFileVersion.V3.getVersionBytes())) {
+ throw new IllegalStateException(
+
StorageEngineMessages.EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4);
+ }
+ return provider;
+ }
+
+ public static WALFileVersion writeVersion(WALFileVersion requested) {
+ return Holder.PROVIDER != null && Holder.PROVIDER.isEnabled()
+ ? WALFileVersion.EXTENDED
+ : requested;
+ }
+
+ static boolean recognizes(FileChannel channel) throws IOException {
+ return recognizes(channel, Holder.PROVIDER);
+ }
+
+ static boolean recognizes(FileChannel channel, WALFormatProvider provider)
throws IOException {
+ long position = channel.position();
+ try {
+ if (provider != null) {
+ byte[] magic = provider.getMagic();
+ if (channel.size() >= magic.length) {
+ ByteBuffer bytes = ByteBuffer.allocate(magic.length);
+ IOUtils.readFully(channel, bytes, 0);
+ if (Arrays.equals(magic, bytes.array())) {
+ return true;
+ }
+ }
+ }
+ // Versioned magic is reserved: an unavailable future format must not
fall back to V1.
+ int size = (int) Math.min(channel.size(), 64);
+ ByteBuffer prefix = ByteBuffer.allocate(size);
+ IOUtils.readFully(channel, prefix, 0);
+ if (new String(prefix.array(),
StandardCharsets.US_ASCII).matches("(?s)^V[0-9]+-WAL.*")) {
+ throw WALFormatException.unsupported();
+ }
+ return false;
+ } finally {
+ channel.position(position);
+ }
+ }
+
+ public static WALFileContext create(File file, FileChannel channel) throws
IOException {
+ if (Holder.PROVIDER == null || !Holder.PROVIDER.isEnabled()) {
+ throw WALFormatException.unsupported();
+ }
+ return requireContext(Holder.PROVIDER.create(file, channel));
+ }
+
+ public static WALFileContext open(File file, FileChannel channel) throws
IOException {
+ if (Holder.PROVIDER == null) {
+ throw WALFormatException.unsupported();
+ }
+ return requireContext(Holder.PROVIDER.open(file, channel));
+ }
+
+ private static WALFileContext requireContext(WALFileContext context) throws
WALFormatException {
+ if (context == null) {
+ throw WALFormatException.unsupported();
+ }
+ return context;
+ }
+
+ public static void validateRead(File file, WALFileVersion version) throws
IOException {
+ if (Holder.PROVIDER != null) {
+ Holder.PROVIDER.validateRead(file, version);
+ }
+ }
+
+ public static WALMetaData readMetadata(File file, FileChannel channel)
throws IOException {
+ if (Holder.PROVIDER == null) {
+ throw WALFormatException.unsupported();
+ }
+ Holder.PROVIDER.validateRead(file, WALFileVersion.EXTENDED);
+ return Holder.PROVIDER.readMetadata(file, channel);
+ }
+
+ public static void validateForRecovery(File file) throws IOException {
+ try (FileChannel channel = FileChannel.open(file.toPath())) {
+ WALFileVersion version = WALFileVersion.getVersion(channel);
+ validateRead(file, version);
+ if (version == WALFileVersion.EXTENDED) {
+ Holder.PROVIDER.validateForRecovery(file);
+ }
+ }
+ }
+
+ public static void onRecoveryFailure(Throwable failure) {
+ if (Holder.PROVIDER != null) {
+ Holder.PROVIDER.onRecoveryFailure(failure);
+ }
+ }
+}
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 d4b4a8b78f9..7a512ff5a95 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
@@ -53,6 +53,7 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
private ByteBuffer dataBuffer = null;
private ByteBuffer compressedBuffer = null;
private long currentSegmentOffset = -1;
+ private long currentSegmentBlockIndex;
private long suspendedSegmentOffset = -1;
private int suspendedSegmentPosition;
private int suspendedSegmentRemaining;
@@ -67,6 +68,9 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
private long endOffset = -1;
private long logicalReadOffset;
private boolean recoveringEntries;
+ private WALFileContext fileContext;
+ private long dataStartOffset;
+ private long blockIndex;
WALFileVersion version;
@@ -81,7 +85,12 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
try {
fileSize = channel.size();
analyzeFileVersion();
- if (ignoreMetadata) {
+ WALFormats.validateRead(logFile, version);
+ if (version == WALFileVersion.EXTENDED) {
+ fileContext = WALFormats.open(logFile, channel);
+ dataStartOffset = channel.position();
+ endOffset = fileSize;
+ } else 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;
@@ -89,8 +98,23 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
} else {
getEndOffset();
}
- } catch (Exception e) {
- channel.close();
+ } catch (IOException | RuntimeException | Error e) {
+ try {
+ channel.close();
+ } catch (IOException closeException) {
+ e.addSuppressed(closeException);
+ }
+ if (fileContext != null) {
+ try {
+ fileContext.close();
+ } catch (RuntimeException | Error closeException) {
+ e.addSuppressed(closeException);
+ }
+ }
+ if (version == WALFileVersion.EXTENDED &&
!WALFormatException.isFatal(e)) {
+ throw new WALFormatException(
+
StorageEngineMessages.EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4,
e);
+ }
throw e;
}
}
@@ -212,9 +236,18 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
try {
channel.close();
} finally {
- releaseReadBuffers();
- suspendedSegmentOffset = -1;
- suspendedSegmentRemaining = 0;
+ try {
+ releaseReadBuffers();
+ } finally {
+ try {
+ if (fileContext != null) {
+ fileContext.close();
+ }
+ } finally {
+ suspendedSegmentOffset = -1;
+ suspendedSegmentRemaining = 0;
+ }
+ }
}
}
@@ -229,7 +262,7 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
suspendedSegmentPosition = dataBuffer.position();
suspendedSegmentRemaining = dataBuffer.remaining();
}
- MmapUtil.clean(dataBuffer);
+ releaseSegment();
MmapUtil.clean(compressedBuffer);
dataBuffer = null;
compressedBuffer = null;
@@ -249,13 +282,18 @@ public class WALInputStream extends InputStream
implements AutoCloseable {
private void loadNextSegment() throws IOException {
if (suspendedSegmentOffset >= 0) {
channel.position(suspendedSegmentOffset);
+ if (version == WALFileVersion.EXTENDED) {
+ blockIndex = currentSegmentBlockIndex;
+ }
}
if (channel.position() >= endOffset) {
throw new
EOFException(StorageEngineMessages.REACH_END_OFFSET_OF_WAL_FILE);
}
long startTime = System.nanoTime();
long startPosition = channel.position();
- if (version == WALFileVersion.V2 || version == WALFileVersion.V3) {
+ if (version == WALFileVersion.EXTENDED) {
+ loadNextExtendedSegment();
+ } else if (version == WALFileVersion.V2 || version == WALFileVersion.V3) {
loadNextSegmentV2();
} else if (version == WALFileVersion.V1) {
loadNextSegmentV1();
@@ -275,6 +313,30 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
.recordWALRead(channel.position() - startPosition, System.nanoTime() -
startTime);
}
+ private void loadNextExtendedSegment() throws IOException {
+ releaseSegment();
+ currentSegmentOffset = channel.position();
+ currentSegmentBlockIndex = blockIndex;
+ dataBuffer = fileContext.readSegment(channel, endOffset, blockIndex);
+ if (dataBuffer == null || !dataBuffer.hasRemaining()) {
+ throw WALFormatException.unsupported();
+ }
+ blockIndex++;
+ }
+
+ private void releaseSegment() {
+ ByteBuffer segment = dataBuffer;
+ dataBuffer = null;
+ if (segment == null) {
+ return;
+ }
+ if (fileContext != null) {
+ fileContext.release(segment);
+ } else {
+ MmapUtil.clean(segment);
+ }
+ }
+
private void loadNextSegmentV1() throws IOException {
// just read raw data as input
if (channel.position() >= fileSize) {
@@ -389,6 +451,25 @@ public class WALInputStream extends InputStream implements
AutoCloseable {
releaseReadBuffers();
suspendedSegmentOffset = -1;
suspendedSegmentRemaining = 0;
+ if (version == WALFileVersion.EXTENDED) {
+ if (pos < 0) {
+ throw new WALFormatException(
+
StorageEngineMessages.EXCEPTION_INVALID_WAL_LOGICAL_POSITION_8CF34AFE);
+ }
+ channel.position(dataStartOffset);
+ blockIndex = 0;
+ logicalReadOffset = 0;
+ long remaining = pos;
+ while (true) {
+ loadNextExtendedSegment();
+ if (remaining < dataBuffer.remaining()) {
+ dataBuffer.position((int) remaining);
+ logicalReadOffset = pos;
+ return;
+ }
+ remaining -= dataBuffer.remaining();
+ }
+ }
if (version == WALFileVersion.V2 || version == WALFileVersion.V3) {
channel.position(version.getVersionBytes().length);
long posRemain = pos;
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 5d316a7508f..0cf97ba8809 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
@@ -412,6 +412,9 @@ public class WALMetaData implements SerializedSize {
try {
return readFromWALFileWithoutRecovery(logFile, channel);
} catch (IOException metadataFailure) {
+ if (WALFormatException.isFatal(metadataFailure)) {
+ throw 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
@@ -457,6 +460,9 @@ public class WALMetaData implements SerializedSize {
return new WALMetaData();
}
WALFileVersion version = WALFileVersion.getVersion(channel);
+ if (version == WALFileVersion.EXTENDED) {
+ return WALFormats.readMetadata(logFile, channel);
+ }
if (channel.size() < version.getVersionBytes().length + Integer.BYTES
|| !isValidMagicString(channel, version)) {
throw new BrokenWALFileException(logFile);
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 c54f3e45bc4..a214b25211c 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
@@ -32,6 +32,7 @@ import java.io.DataInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.Iterator;
import java.util.NoSuchElementException;
@@ -85,16 +86,24 @@ public class WALReader implements Closeable {
nextEntry = null;
return false;
}
+ } catch (WALFormatException e) {
+ throw new UncheckedIOException(e);
} catch (EOFException e) {
fileCorrupted = true;
return false;
} catch (IOException e) {
+ if (WALFormatException.isFatal(e)) {
+ throw new UncheckedIOException(e);
+ }
fileCorrupted = true;
// log only when file should be complete
if (!fileMayCorrupt) {
logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_ENTRY_SKIP_BROKEN,
logFile, e);
}
} catch (Exception e) {
+ if (WALFormatException.isFatal(e)) {
+ throw new IllegalStateException(e);
+ }
fileCorrupted = true;
logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_ENTRY_SKIP_BROKEN,
logFile, e);
}
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 d340b7ce0df..2c3a1c27acb 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
@@ -42,8 +42,8 @@ public class WALWriter extends LogWriter {
}
public WALWriter(File logFile, WALFileVersion version) throws IOException {
- super(logFile, version);
- this.version = version;
+ super(logFile, WALFormats.writeVersion(version));
+ this.version = WALFormats.writeVersion(version);
}
/**
@@ -68,7 +68,11 @@ public class WALWriter extends LogWriter {
private synchronized void endFile() throws IOException {
// The writer channel is write-only; its known version and size identify
an empty header.
- if (version != WALFileVersion.V1 && logChannel.size() ==
version.getVersionBytes().length) {
+ if (version != WALFileVersion.V1
+ && logChannel.size()
+ == (version == WALFileVersion.EXTENDED
+ ? dataStartOffset
+ : version.getVersionBytes().length)) {
// A WAL with no entries is valid and has no marker or metadata trailer
to append.
super.close();
return;
@@ -78,6 +82,10 @@ public class WALWriter extends LogWriter {
// mark info part ends
endMarker.serialize(markerBuffer);
write(markerBuffer, false);
+ if (fileContext != null) {
+ fileContext.seal(logChannel, metaData, blockIndex++);
+ return;
+ }
int metaDataSize = metaData.serializedSize(version);
ByteBuffer buffer =
@@ -100,9 +108,12 @@ public class WALWriter extends LogWriter {
@Override
public void close() throws IOException {
+ if (!logChannel.isOpen()) {
+ return;
+ }
try {
endFile();
- } catch (IOException | RuntimeException e) {
+ } catch (IOException | RuntimeException | Error 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 {
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 73d429e3ccb..180890e0738 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
@@ -33,6 +33,8 @@ import
org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
import
org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.MemTableInfo;
import
org.apache.iotdb.db.storageengine.dataregion.wal.exception.BrokenWALFileException;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALByteBufReader;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormatException;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormats;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALReader;
import
org.apache.iotdb.db.storageengine.dataregion.wal.recover.file.UnsealedTsFileRecoverPerformer;
@@ -47,6 +49,7 @@ import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.File;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.ArrayList;
@@ -72,7 +75,7 @@ public class WALNodeRecoverTask implements Runnable {
// version id of first valid .wal file
private long firstValidVersionId = Long.MAX_VALUE;
private Map<Long, MemTableInfo> memTableId2Info;
- private Map<Long, UnsealedTsFileRecoverPerformer>
memTableId2RecoverPerformer;
+ private Map<Long, UnsealedTsFileRecoverPerformer>
memTableId2RecoverPerformer = new HashMap<>();
public WALNodeRecoverTask(File logDirectory, CountDownLatch
allNodesRecoveredLatch) {
this.logDirectory = logDirectory;
@@ -81,6 +84,32 @@ public class WALNodeRecoverTask implements Runnable {
@Override
public void run() {
+ try {
+ validateFilesBeforeRecovery();
+ recoverNode();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ } finally {
+ allNodesRecoveredLatch.countDown();
+ }
+ }
+
+ private void validateFilesBeforeRecovery() throws IOException {
+ File[] walFiles = WALFileUtils.listAllWALFiles(logDirectory);
+ File[] checkpoints =
CheckpointFileUtils.listAllCheckpointFiles(logDirectory);
+ if (walFiles != null) {
+ for (File file : walFiles) {
+ WALFormats.validateForRecovery(file);
+ }
+ }
+ if (checkpoints != null) {
+ for (File file : checkpoints) {
+ WALFormats.validateForRecovery(file);
+ }
+ }
+ }
+
+ private void recoverNode() {
logger.info(StorageEngineMessages.START_RECOVERING_WAL_NODE_IN_DIR,
logDirectory);
// recover version id and search index
@@ -95,6 +124,9 @@ public class WALNodeRecoverTask implements Runnable {
for (UnsealedTsFileRecoverPerformer recoverPerformer :
memTableId2RecoverPerformer.values()) {
recoverPerformer.getRecoverListener().fail(e);
}
+ if (WALFormatException.isFatal(e)) {
+ throw new IllegalStateException(e);
+ }
} finally {
for (UnsealedTsFileRecoverPerformer recoverPerformer :
memTableId2RecoverPerformer.values()) {
try {
@@ -107,53 +139,49 @@ public class WALNodeRecoverTask implements Runnable {
}
}
- try {
- if
(config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS))
{
- // delete checkpoint info to avoid repeated recover
- File[] checkpointFiles =
CheckpointFileUtils.listAllCheckpointFiles(logDirectory);
- for (File checkpointFile : checkpointFiles) {
- try {
- Files.delete(checkpointFile.toPath());
- } catch (IOException e) {
- logger.error(StorageEngineMessages.ERROR_DELETE_CHECKPOINT_FILE,
checkpointFile, e);
- }
- }
- // register wal node
- WALManager.getInstance()
- .registerWALNode(
- logDirectory.getName(),
- logDirectory.getAbsolutePath(),
- lastVersionId + 1,
- lastSearchIndex);
- logger.info(
- StorageEngineMessages
-
.STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_ADD_THIS_FA6ADE22,
- logDirectory);
- } else {
- if (cleanupRecoveredDirectory(logDirectory)) {
- logger.info(
- StorageEngineMessages
-
.STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_SO_DELETE_A17892D9,
- logDirectory);
+ if
(config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS))
{
+ // delete checkpoint info to avoid repeated recover
+ File[] checkpointFiles =
CheckpointFileUtils.listAllCheckpointFiles(logDirectory);
+ for (File checkpointFile : checkpointFiles) {
+ try {
+ Files.delete(checkpointFile.toPath());
+ } catch (IOException e) {
+ logger.error(StorageEngineMessages.ERROR_DELETE_CHECKPOINT_FILE,
checkpointFile, e);
}
}
-
- // IoTConsensusV2 will not only delete WAL node folder, but also
register WAL node.
- if
(config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS_V2))
{
- // register wal node
- WALManager.getInstance()
- .registerWALNode(
- logDirectory.getName(),
- logDirectory.getAbsolutePath(),
- lastVersionId + 1,
- lastSearchIndex);
+ // register wal node
+ WALManager.getInstance()
+ .registerWALNode(
+ logDirectory.getName(),
+ logDirectory.getAbsolutePath(),
+ lastVersionId + 1,
+ lastSearchIndex);
+ logger.info(
+ StorageEngineMessages
+
.STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_ADD_THIS_FA6ADE22,
+ logDirectory);
+ } else {
+ if (cleanupRecoveredDirectory(logDirectory)) {
logger.info(
StorageEngineMessages
-
.STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_ADD_THIS_FA6ADE22,
+
.STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_SO_DELETE_A17892D9,
logDirectory);
}
- } finally {
- allNodesRecoveredLatch.countDown();
+ }
+
+ // IoTConsensusV2 will not only delete WAL node folder, but also register
WAL node.
+ if
(config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS_V2))
{
+ // register wal node
+ WALManager.getInstance()
+ .registerWALNode(
+ logDirectory.getName(),
+ logDirectory.getAbsolutePath(),
+ lastVersionId + 1,
+ lastSearchIndex);
+ logger.info(
+ StorageEngineMessages
+
.STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_ADD_THIS_FA6ADE22,
+ logDirectory);
}
}
@@ -192,6 +220,9 @@ public class WALNodeRecoverTask implements Runnable {
previousLogicalOffset = logicalOffset;
}
} catch (Exception e) {
+ if (WALFormatException.isFatal(e)) {
+ throw new IllegalStateException(e);
+ }
logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_LOGS_SKIP,
lastWALFile, e);
}
// make sure last wal file is correct
@@ -241,6 +272,9 @@ public class WALNodeRecoverTask implements Runnable {
try {
return walRepairWriter.repair(metaData);
} catch (IOException e) {
+ if (WALFormatException.isFatal(e)) {
+ throw new IllegalStateException(e);
+ }
logger.error(StorageEngineMessages.FAIL_TO_RECOVER_WAL_METADATA,
walFile, e);
DataNodeExceptionMetrics.getInstance().recordSuspiciousDiskException(e);
return false;
@@ -284,6 +318,9 @@ public class WALNodeRecoverTask implements Runnable {
recoverPerformer.startRecovery();
} catch (Exception e) {
recoverPerformer.getRecoverListener().fail(e);
+ if (WALFormatException.isFatal(e)) {
+ throw new IllegalStateException(e);
+ }
}
}
// find all valid .wal files
@@ -335,12 +372,18 @@ public class WALNodeRecoverTask implements Runnable {
logDirectory.getName(),
e);
} catch (IOException e) {
+ if (WALFormatException.isFatal(e)) {
+ throw new IllegalStateException(e);
+ }
logger.warn(
StorageEngineMessages
.STORAGE_LOG_FAIL_TO_READ_MEMTABLE_IDS_FROM_THE_WAL_FILE_OF_WAL_NODE_FBCE8D93,
walFile.getAbsoluteFile(),
e);
} catch (Exception e) {
+ if (WALFormatException.isFatal(e)) {
+ throw new IllegalStateException(e);
+ }
logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_LOGS_SKIP, walFile,
e);
}
}
@@ -355,6 +398,9 @@ public class WALNodeRecoverTask implements Runnable {
recoverPerformer.getRecoverListener().succeed();
} catch (Exception e) {
recoverPerformer.getRecoverListener().fail(e);
+ if (WALFormatException.isFatal(e)) {
+ throw new IllegalStateException(e);
+ }
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java
index 54a45626a3d..c6ae45b497c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java
@@ -31,6 +31,8 @@ import org.apache.iotdb.db.exception.DataRegionException;
import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException;
import org.apache.iotdb.db.i18n.StorageEngineMessages;
import
org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALRecoverException;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormatException;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormats;
import
org.apache.iotdb.db.storageengine.dataregion.wal.recover.file.UnsealedTsFileRecoverPerformer;
import
org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALRecoverListener;
@@ -108,12 +110,32 @@ public class WALRecoverManager {
recoverThreadPool =
IoTDBThreadPoolFactory.newCachedThreadPool(ThreadName.WAL_RECOVER.getName());
CountDownLatch allNodesRecoveredLatch = new
CountDownLatch(walNodeDirs.size());
+ List<Future<?>> recoveries = new ArrayList<>();
for (File walNodeDir : walNodeDirs) {
- recoverThreadPool.submit(new WALNodeRecoverTask(walNodeDir,
allNodesRecoveredLatch));
+ recoveries.add(
+ recoverThreadPool.submit(new WALNodeRecoverTask(walNodeDir,
allNodesRecoveredLatch)));
}
try {
allNodesRecoveredLatch.await();
+ ExecutionException failure = null;
+ for (Future<?> recovery : recoveries) {
+ try {
+ recovery.get();
+ } catch (ExecutionException e) {
+ if (failure == null) {
+ failure = e;
+ } else if (WALFormatException.isFatal(e) &&
!WALFormatException.isFatal(failure)) {
+ e.addSuppressed(failure);
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new
WALRecoverException(StorageEngineMessages.FAIL_TO_RECOVER_WAL, e);
@@ -128,6 +150,10 @@ public class WALRecoverManager {
absolutePath2RecoverPerformer.values()) {
recoverPerformer.getRecoverListener().fail(e);
}
+ if (WALFormatException.isFatal(e)) {
+ WALFormats.onRecoveryFailure(e);
+ throw new
WALRecoverException(StorageEngineMessages.FAIL_TO_RECOVER_WAL, e);
+ }
} finally {
for (UnsealedTsFileRecoverPerformer recoverPerformer :
absolutePath2RecoverPerformer.values()) {
@@ -169,25 +195,31 @@ public class WALRecoverManager {
recoverPerformer.getTsFileAbsolutePath(),
e);
recoverPerformer.getRecoverListener().fail(e);
+ if (WALFormatException.isFatal(e)) {
+ throw e;
+ }
}
return null;
};
futures.add(recoverTsFilesThreadPool.submit(recoverTsFileTask));
}
// wait until all tasks done
- for (Future<Void> future : futures) {
- try {
- future.get();
- } catch (ExecutionException e) {
- throw new StorageEngineFailureException(
- StorageEngineMessages.STORAGE_ENGINE_FAILED_TO_RECOVER, e);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new StorageEngineFailureException(
- StorageEngineMessages.STORAGE_ENGINE_FAILED_TO_RECOVER, e);
+ try {
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (ExecutionException e) {
+ throw new StorageEngineFailureException(
+ StorageEngineMessages.STORAGE_ENGINE_FAILED_TO_RECOVER, e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new StorageEngineFailureException(
+ StorageEngineMessages.STORAGE_ENGINE_FAILED_TO_RECOVER, e);
+ }
}
+ } finally {
+ recoverTsFilesThreadPool.shutdownNow();
}
- recoverTsFilesThreadPool.shutdown();
}
public WALRecoverListener addRecoverPerformer(UnsealedTsFileRecoverPerformer
recoverPerformer) {
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 7572eac68cb..962c4d5077f 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
@@ -23,6 +23,7 @@ import org.apache.iotdb.db.i18n.StorageEngineMessages;
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.WALFormatException;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter;
@@ -146,6 +147,9 @@ public class WALRepairWriter {
WALMetaData.readFromWALFileWithoutRecovery(logFile, channel);
return true;
} catch (IOException | RuntimeException e) {
+ if (WALFormatException.isFatal(e)) {
+ throw new IllegalStateException(e);
+ }
return false;
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALFileUtils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALFileUtils.java
index 5cff6d4b13c..abb137448b8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALFileUtils.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALFileUtils.java
@@ -339,11 +339,11 @@ public class WALFileUtils {
for (final File walFile : walFiles) {
try (final FileChannel channel = FileChannel.open(walFile.toPath())) {
- if (WALFileVersion.getVersion(channel) != WALFileVersion.V3) {
+ WALFileVersion version = WALFileVersion.getVersion(channel);
+ if (version != WALFileVersion.V3 && version !=
WALFileVersion.EXTENDED) {
continue;
}
- // V3 WAL stores writer progress in the footer, so these lookups do
not deserialize entry
- // bodies.
+ // Formats with writer progress in the footer avoid deserializing
entry bodies here.
final WALMetaData metadata = WALMetaData.readFromWALFile(walFile,
channel);
final List<Long> physicalTimes = metadata.getPhysicalTimes();
final List<Short> nodeIds = metadata.getNodeIds();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatExtensionTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatExtensionTest.java
new file mode 100644
index 00000000000..3ead148f06a
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatExtensionTest.java
@@ -0,0 +1,649 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+
+import org.apache.iotdb.commons.concurrent.ExceptionalCountDownLatch;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.exception.StartupException;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.storageengine.StorageEngine;
+import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.Checkpoint;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.CheckpointType;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.MemTableInfo;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALRecoverException;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALNodeRecoverTask;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRepairWriter;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.recover.file.UnsealedTsFileRecoverPerformer;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALRecoverListener;
+import org.apache.iotdb.db.utils.MmapUtil;
+
+import io.netty.util.internal.PlatformDependent;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.fileSystem.FSFactoryProducer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.mockito.ArgumentCaptor;
+
+import java.io.EOFException;
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Pattern;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Exercises dispatch with a test-only provider; it implements no encryption
or product format. */
+public class WALFormatExtensionTest {
+ @Rule public final TemporaryFolder folder = new TemporaryFolder();
+
+ @Test
+ public void testExtensionDispatchInIsolatedRuntime() throws Exception {
+ String serviceName = "META-INF/services/" +
WALFormatProvider.class.getName();
+ File serviceFile = new File(folder.getRoot(), serviceName);
+ Files.createDirectories(serviceFile.toPath().getParent());
+ Files.writeString(serviceFile.toPath(), StubProvider.class.getName());
+ URL service = serviceFile.toURI().toURL();
+ String[] classpath =
+ System.getProperty("surefire.test.class.path",
System.getProperty("java.class.path"))
+ .split(Pattern.quote(File.pathSeparator));
+ URL[] urls = new URL[classpath.length];
+ for (int i = 0; i < classpath.length; i++) {
+ urls[i] = new File(classpath[i]).toURI().toURL();
+ }
+ // Do not register a global test provider: downstream editions install
their own provider.
+ ClassLoader previous = Thread.currentThread().getContextClassLoader();
+ try (URLClassLoader isolated =
+ new URLClassLoader(urls, ClassLoader.getPlatformClassLoader()) {
+ @Override
+ public Enumeration<URL> getResources(String name) throws IOException
{
+ return serviceName.equals(name)
+ ? Collections.enumeration(Collections.singleton(service))
+ : super.getResources(name);
+ }
+ }) {
+ Thread.currentThread().setContextClassLoader(isolated);
+ Class<?> cases = isolated.loadClass(Cases.class.getName());
+ Object result =
+ isolated
+ .loadClass("org.junit.runner.JUnitCore")
+ .getMethod("runClasses", Class[].class)
+ .invoke(null, (Object) new Class<?>[] {cases});
+ assertTrue(
+ result.getClass().getMethod("getFailures").invoke(result).toString(),
+ (boolean)
result.getClass().getMethod("wasSuccessful").invoke(result));
+ assertEquals(16,
result.getClass().getMethod("getRunCount").invoke(result));
+ } finally {
+ Thread.currentThread().setContextClassLoader(previous);
+ }
+ }
+
+ public static class Cases {
+ @Rule public final TemporaryFolder folder = new TemporaryFolder();
+ private WALFormatProvider provider;
+ private WALFileContext context;
+
+ @Before
+ public void setUp() throws Exception {
+ provider = mock(WALFormatProvider.class);
+ context = mock(WALFileContext.class);
+ when(provider.isEnabled()).thenReturn(true);
+ when(provider.create(any(), any())).thenReturn(context);
+ when(provider.open(any(), any())).thenReturn(context);
+ StubProvider.delegate = provider;
+ }
+
+ @After
+ public void tearDown() {
+ StubProvider.delegate = null;
+ }
+
+ @Test
+ public void testInitializationFailureClosesCallerChannels() throws
Exception {
+ File file = folder.newFile();
+ WALFormatException failure = WALFormatException.unsupported();
+ when(provider.create(any(), any())).thenThrow(failure);
+ assertSame(failure, assertThrows(IOException.class, () -> new
WALWriter(file)));
+ ArgumentCaptor<FileChannel> created =
ArgumentCaptor.forClass(FileChannel.class);
+ verify(provider).create(eq(file), created.capture());
+ assertFalse(created.getValue().isOpen());
+ when(provider.open(any(), any())).thenThrow(failure);
+ assertSame(failure, assertThrows(IOException.class, () -> new
WALInputStream(file)));
+ ArgumentCaptor<FileChannel> opened =
ArgumentCaptor.forClass(FileChannel.class);
+ verify(provider).open(eq(file), opened.capture());
+ assertFalse(opened.getValue().isOpen());
+ }
+
+ @Test
+ public void testWriterDelegatesSegmentsMetadataAndClose() throws Exception
{
+ when(context.writeData(any(), anyLong(), any(), anyInt(), any()))
+ .thenAnswer(
+ invocation -> {
+ FileChannel channel = invocation.getArgument(0);
+ ByteBuffer bytes = invocation.getArgument(4);
+ int size = bytes.remaining();
+ while (bytes.hasRemaining()) {
+ channel.write(bytes);
+ }
+ return size;
+ });
+ File file = folder.newFile();
+ try (WALWriter writer = new WALWriter(file)) {
+ assertEquals(WALFileVersion.EXTENDED, WALFileVersion.getVersion(file));
+ ByteBuffer bytes = ByteBuffer.allocate(Integer.BYTES).putInt(42);
+ writer.write(bytes, false);
+ }
+ ArgumentCaptor<Long> indexes = ArgumentCaptor.forClass(Long.class);
+ verify(context, times(2))
+ .writeData(any(), indexes.capture(),
eq(CompressionType.UNCOMPRESSED), anyInt(), any());
+ assertEquals(Arrays.asList(0L, 1L), indexes.getAllValues());
+ verify(context).seal(any(), any(), eq(2L));
+ verify(context).close();
+ }
+
+ @Test
+ public void testCheckpointUsesExtensionWithoutWalMetadata() throws
Exception {
+ try (CheckpointWriter writer = new CheckpointWriter(folder.newFile())) {
+ writer.write(ByteBuffer.allocate(Long.BYTES).putLong(7), false);
+ }
+ verify(context).writeData(any(), eq(0L),
eq(CompressionType.UNCOMPRESSED), eq(8), any());
+ verify(context, never()).seal(any(), any(), anyLong());
+ verify(context).close();
+ }
+
+ @Test
+ public void testReaderResumesAndSeeksThroughProvider() throws Exception {
+ File file = extensionFile();
+ when(context.readSegment(any(), anyLong(), anyLong()))
+ .thenAnswer(
+ invocation -> {
+ if ((long) invocation.getArgument(2) != 0) {
+ throw new EOFException();
+ }
+ FileChannel channel = invocation.getArgument(0);
+ ByteBuffer bytes = ByteBuffer.allocate(4);
+ while (bytes.hasRemaining()) {
+ if (channel.read(bytes) < 0) {
+ throw new EOFException();
+ }
+ }
+ return bytes.flip();
+ });
+ doAnswer(
+ invocation -> {
+ ByteBuffer bytes = invocation.getArgument(0);
+ if (bytes != null) {
+ Arrays.fill(bytes.array(), (byte) 0);
+ }
+ return null;
+ })
+ .when(context)
+ .release(any());
+ try (WALInputStream reader = new WALInputStream(file)) {
+ assertEquals(1, reader.read());
+ reader.releaseReadBuffers();
+ assertEquals(2, reader.read());
+ reader.skipToGivenLogicalPosition(3);
+ assertEquals(4, reader.read());
+ }
+ verify(context, times(3)).readSegment(any(), anyLong(), eq(0L));
+ verify(context).close();
+ }
+
+ @Test
+ public void testFatalMetadataCannotTriggerRepairOrQuarantine() throws
Exception {
+ File file = extensionFile();
+ byte[] original = Files.readAllBytes(file.toPath());
+ WALFormatException failure = WALFormatException.unsupported();
+ when(provider.readMetadata(any(), any())).thenThrow(failure);
+ try (FileChannel channel = FileChannel.open(file.toPath())) {
+ assertSame(
+ failure,
+ assertThrows(IOException.class, () ->
WALMetaData.readFromWALFile(file, channel)));
+ }
+ IllegalStateException repairFailure =
+ assertThrows(
+ IllegalStateException.class,
+ () -> new WALRepairWriter(file).repair(new WALMetaData()));
+ assertTrue(WALFormatException.isFatal(repairFailure));
+ assertArrayEquals(original, Files.readAllBytes(file.toPath()));
+ assertFalse(new File(file + ".broken").exists());
+ verify(provider, never()).open(any(), any());
+ }
+
+ @Test
+ public void testFatalSegmentFailureIsNotSwallowedByReaders() throws
Exception {
+ File file = extensionFile();
+ when(context.readSegment(any(), anyLong(), anyLong()))
+ .thenThrow(new IOException(WALFormatException.unsupported()));
+ try (WALReader reader = new WALReader(file, true)) {
+ assertTrue(
+
WALFormatException.isFatal(assertThrows(UncheckedIOException.class,
reader::hasNext)));
+ }
+ assertTrue(
+ WALFormatException.isFatal(
+ assertThrows(UncheckedIOException.class, () -> new
CheckpointReader(file))));
+ }
+
+ @Test
+ public void testRecoveryPreflightFailureReachesCallerAndKeepsFiles()
throws Exception {
+ File walRoot = folder.newFolder("wal");
+ File node = new File(walRoot, "node");
+ assertTrue(node.mkdir());
+ File wal = new File(node, "_0-0-1.wal");
+ Files.write(wal.toPath(), StubProvider.MAGIC);
+ byte[] original = Files.readAllBytes(wal.toPath());
+
doThrow(WALFormatException.unsupported()).when(provider).validateForRecovery(any());
+ String[] previous =
CommonDescriptor.getInstance().getConfig().getWalDirs();
+ WALRecoverManager recovery = WALRecoverManager.getInstance();
+ try {
+ recovery.clear();
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setWalDirs(new String[] {walRoot.getAbsolutePath()});
+ recovery.setAllDataRegionScannedLatch(new
ExceptionalCountDownLatch(0));
+ assertTrue(
+ WALFormatException.isFatal(assertThrows(WALRecoverException.class,
recovery::recover)));
+ assertArrayEquals(original, Files.readAllBytes(wal.toPath()));
+ verify(provider).onRecoveryFailure(any());
+ } finally {
+ recovery.clear();
+ CommonDescriptor.getInstance().getConfig().setWalDirs(previous);
+ }
+ }
+
+ @Test
+ public void testOrdinaryFailureDoesNotHideLaterFatalFailure() throws
Exception {
+ assertMultipleNodeFailures(false, true);
+ }
+
+ @Test
+ public void testFatalFailureRetainsLaterOrdinaryFailure() throws Exception
{
+ assertMultipleNodeFailures(true, true);
+ }
+
+ @Test
+ public void testOrdinaryNodeFailuresKeepExistingRecoveryPolicy() throws
Exception {
+ assertMultipleNodeFailures(false, false);
+ }
+
+ private void assertMultipleNodeFailures(boolean fatalFirst, boolean
hasFatal) throws Exception {
+ File firstRoot = folder.newFolder("first-wal");
+ File secondRoot = folder.newFolder("second-wal");
+ File[] roots = {firstRoot, secondRoot};
+ IOException ordinary = new IOException("ordinary recovery failure");
+ IOException other = hasFatal ? WALFormatException.unsupported() : new
IOException("other");
+ IOException[] failures =
+ fatalFirst ? new IOException[] {other, ordinary} : new IOException[]
{ordinary, other};
+ List<File> files = new ArrayList<>();
+ for (int i = 0; i < roots.length; i++) {
+ File node = new File(roots[i], "node");
+ assertTrue(node.mkdir());
+ File wal = new File(node, "_0-0-1.wal");
+ Files.write(wal.toPath(), StubProvider.MAGIC);
+ files.add(wal);
+ doThrow(failures[i]).when(provider).validateForRecovery(wal);
+ }
+ String[] previous =
CommonDescriptor.getInstance().getConfig().getWalDirs();
+ WALRecoverManager recovery = WALRecoverManager.getInstance();
+ try {
+ recovery.clear();
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setWalDirs(new String[] {firstRoot.getAbsolutePath(),
secondRoot.getAbsolutePath()});
+ recovery.setAllDataRegionScannedLatch(new
ExceptionalCountDownLatch(0));
+ if (hasFatal) {
+ WALRecoverException thrown = assertThrows(WALRecoverException.class,
recovery::recover);
+ assertTrue(WALFormatException.isFatal(thrown));
+ Throwable primary = thrown.getCause();
+ assertSame(other, primary.getCause().getCause());
+ assertEquals(1, primary.getSuppressed().length);
+ assertSame(ordinary,
primary.getSuppressed()[0].getCause().getCause());
+ verify(provider).onRecoveryFailure(primary);
+ } else {
+ recovery.recover();
+ verify(provider, never()).onRecoveryFailure(any());
+ }
+ for (File wal : files) {
+ verify(provider).validateForRecovery(wal);
+ assertArrayEquals(StubProvider.MAGIC,
Files.readAllBytes(wal.toPath()));
+ }
+ } finally {
+ recovery.clear();
+ CommonDescriptor.getInstance().getConfig().setWalDirs(previous);
+ }
+ }
+
+ private File extensionFile() throws IOException {
+ File file = folder.newFile();
+ ByteBuffer bytes = ByteBuffer.allocate(StubProvider.MAGIC.length + 4);
+ bytes.put(StubProvider.MAGIC).put(new byte[] {1, 2, 3, 4});
+ Files.write(file.toPath(), bytes.array());
+ return file;
+ }
+
+ @Test
+ public void testProviderOwnsPooledDirectBufferAcrossPauseSeekAndClose()
throws Exception {
+ File file = extensionFile();
+ ByteBuffer pooled = ByteBuffer.allocateDirect(4);
+ long address = PlatformDependent.directBufferAddress(pooled);
+ assertTrue(address != 0);
+ when(context.readSegment(any(), anyLong(), anyLong()))
+ .thenAnswer(
+ call -> {
+ FileChannel channel = call.getArgument(0);
+ pooled.clear();
+ while (pooled.hasRemaining()) {
+ if (channel.read(pooled) < 0) {
+ throw new EOFException();
+ }
+ }
+ return pooled.flip();
+ });
+ try {
+ try (WALInputStream reader = new WALInputStream(file)) {
+ assertEquals(1, reader.read());
+ reader.releaseReadBuffers();
+ assertEquals(address, PlatformDependent.directBufferAddress(pooled));
+ assertEquals(2, reader.read());
+ reader.skipToGivenLogicalPosition(3);
+ assertEquals(address, PlatformDependent.directBufferAddress(pooled));
+ assertEquals(4, reader.read());
+ }
+ assertEquals(address, PlatformDependent.directBufferAddress(pooled));
+ verify(context, times(3)).release(pooled);
+ } finally {
+ if (PlatformDependent.directBufferAddress(pooled) != 0) {
+ MmapUtil.clean(pooled);
+ }
+ }
+ }
+
+ @Test
+ public void testFatalRepairWriteFailureKeepsOriginalFile() throws
Exception {
+ assertFatalRepairFailure(false);
+ }
+
+ @Test
+ public void testFatalRepairSealFailureKeepsOriginalFile() throws Exception
{
+ assertFatalRepairFailure(true);
+ }
+
+ private void assertFatalRepairFailure(boolean seal) throws Exception {
+ File file = extensionFile();
+ byte[] original = Files.readAllBytes(file.toPath());
+ when(provider.readMetadata(any(), any())).thenThrow(new
IOException("missing footer"));
+ when(context.writeData(any(), anyLong(), any(), anyInt(), any()))
+ .thenAnswer(
+ call -> {
+ FileChannel channel = call.getArgument(0);
+ ByteBuffer data = call.getArgument(4);
+ int size = data.remaining();
+ while (data.hasRemaining()) {
+ channel.write(data);
+ }
+ return size;
+ });
+ when(context.readSegment(any(), anyLong(), anyLong()))
+ .thenReturn(ByteBuffer.wrap(new byte[] {1, 2, 3, 4}));
+ WALFormatException failure = WALFormatException.unsupported();
+ if (seal) {
+ doThrow(failure).when(context).seal(any(), any(), anyLong());
+ } else {
+ doThrow(failure).when(context).writeData(any(), anyLong(), any(),
anyInt(), any());
+ }
+ WALMetaData metadata = new WALMetaData();
+ metadata.add(4, 0, 1);
+ Method repair =
+ WALNodeRecoverTask.class.getDeclaredMethod(
+ "repairWalFileIfBroken", File.class, WALMetaData.class);
+ repair.setAccessible(true);
+ InvocationTargetException thrown =
+ assertThrows(InvocationTargetException.class, () ->
repair.invoke(null, file, metadata));
+ assertTrue(WALFormatException.isFatal(thrown.getCause()));
+ assertArrayEquals(original, Files.readAllBytes(file.toPath()));
+ assertFalse(new File(file + ".broken").exists());
+ assertEquals(1, folder.getRoot().listFiles().length);
+ }
+
+ @Test(timeout = 10000)
+ public void testFatalReplayFailureCompletesAllTaskListeners() throws
Exception {
+ assertTaskListenersComplete(false);
+ }
+
+ @Test(timeout = 10000)
+ public void testFatalFinalizationFailureCompletesAllTaskListeners() throws
Exception {
+ assertTaskListenersComplete(true);
+ }
+
+ private void assertTaskListenersComplete(boolean failOnEnd) throws
Exception {
+ WALRecoverManager recovery = WALRecoverManager.getInstance();
+ recovery.clear();
+ File node = folder.newFolder("node");
+ List<UnsealedTsFileRecoverPerformer> performers = new ArrayList<>();
+ List<MemTableInfo> tables = new ArrayList<>();
+ WALFormatException failure = WALFormatException.unsupported();
+ try {
+ for (int i = 1; i <= 2; i++) {
+ File tsFile = new File(folder.getRoot(), "sequence/root.sg/0/0/" + i
+ "-1-0-0.tsfile");
+ UnsealedTsFileRecoverPerformer performer =
mock(UnsealedTsFileRecoverPerformer.class);
+ WALRecoverListener listener = new
WALRecoverListener(tsFile.getAbsolutePath());
+ when(performer.getTsFileResource()).thenReturn(new
TsFileResource(tsFile));
+
when(performer.getTsFileAbsolutePath()).thenReturn(tsFile.getAbsolutePath());
+ when(performer.getRecoverListener()).thenReturn(listener);
+ recovery.addRecoverPerformer(performer);
+ performers.add(performer);
+ IMemTable table = mock(IMemTable.class);
+ when(table.getMemTableId()).thenReturn((long) i);
+ when(table.getDataRegionId()).thenReturn("0");
+ tables.add(new MemTableInfo(table, tsFile.getAbsolutePath(), 0));
+ }
+ Checkpoint checkpoint = new
Checkpoint(CheckpointType.GLOBAL_MEMORY_TABLE_INFO, tables);
+ ByteBuffer bytes = ByteBuffer.allocate(Long.BYTES +
checkpoint.serializedSize());
+ bytes.putLong(2);
+ checkpoint.serialize(bytes);
+ File checkpointFile = new File(node, "_0.checkpoint");
+ ByteBuffer fileBytes = ByteBuffer.allocate(StubProvider.MAGIC.length +
bytes.position());
+ fileBytes.put(StubProvider.MAGIC).put(bytes.array());
+ Files.write(checkpointFile.toPath(), fileBytes.array());
+ when(context.readSegment(any(), anyLong(), anyLong()))
+ .thenAnswer(
+ call -> {
+ FileChannel channel = call.getArgument(0);
+ ByteBuffer data = ByteBuffer.allocate(bytes.capacity());
+ while (data.hasRemaining()) {
+ if (channel.read(data) < 0) {
+ throw new EOFException();
+ }
+ }
+ return data.flip();
+ });
+ if (failOnEnd) {
+ doThrow(new WALRecoverException(failure.getMessage(), failure))
+ .when(performers.get(0))
+ .endRecovery();
+ } else {
+ doThrow(failure).when(performers.get(0)).startRecovery();
+ }
+ CountDownLatch latch = new CountDownLatch(1);
+ assertTrue(
+ WALFormatException.isFatal(
+ assertThrows(
+ IllegalStateException.class, () -> new
WALNodeRecoverTask(node, latch).run())));
+ assertEquals(0, latch.getCount());
+ for (UnsealedTsFileRecoverPerformer performer : performers) {
+ assertEquals(
+ WALRecoverListener.Status.FAILURE,
performer.getRecoverListener().waitForResult());
+ verify(performer).close();
+ }
+ assertTrue(checkpointFile.exists());
+ } finally {
+ recovery.clear();
+ }
+ }
+
+ @Test
+ public void testFatalWalFormatPreventsStorageEngineReadiness() throws
Exception {
+ File walRoot = folder.newFolder("startup-wal");
+ File node = new File(walRoot, "node");
+ assertTrue(node.mkdir());
+ Files.writeString(new File(node, "_0-0-1.wal").toPath(), "V99-WAL");
+ String[] previousDirs =
CommonDescriptor.getInstance().getConfig().getWalDirs();
+ IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ String previousConsensus = config.getDataRegionConsensusProtocolClass();
+ WALRecoverManager recovery = WALRecoverManager.getInstance();
+ StorageEngine engine = mock(StorageEngine.class, CALLS_REAL_METHODS);
+ setField(engine, "fsFactory", FSFactoryProducer.getFSFactory());
+ setField(engine, "isReadyForReadAndWrite", new AtomicBoolean(true));
+ setField(engine, "isReadyForNonReadWriteFunctions", new
AtomicBoolean(true));
+ doReturn(Collections.emptyMap()).when(engine).getLocalDataRegionInfo();
+ try {
+ recovery.clear();
+ CommonDescriptor.getInstance().getConfig().setWalDirs(new String[]
{walRoot.getPath()});
+
config.setDataRegionConsensusProtocolClass(ConsensusFactory.SIMPLE_CONSENSUS);
+ Method recover =
StorageEngine.class.getDeclaredMethod("asyncRecoverDataRegion");
+ recover.setAccessible(true);
+ InvocationTargetException thrown =
+ assertThrows(InvocationTargetException.class, () ->
recover.invoke(engine));
+ assertTrue(thrown.getCause() instanceof StartupException);
+ assertTrue(WALFormatException.isFatal(thrown.getCause()));
+ assertFalse(engine.isReadyForReadAndWrite());
+ assertFalse(engine.isReadyForNonReadWriteFunctions());
+ } finally {
+ Field pool = StorageEngine.class.getDeclaredField("cachedThreadPool");
+ pool.setAccessible(true);
+ ExecutorService executor = (ExecutorService) pool.get(engine);
+ if (executor != null) {
+ executor.shutdownNow();
+ }
+ recovery.clear();
+ CommonDescriptor.getInstance().getConfig().setWalDirs(previousDirs);
+ config.setDataRegionConsensusProtocolClass(previousConsensus);
+ }
+ }
+
+ private static void setField(Object target, String name, Object value)
throws Exception {
+ Field field = StorageEngine.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+ }
+
+ /** Registered only inside the isolated test runtime, never in a
distribution. */
+ public static final class StubProvider implements WALFormatProvider {
+ private static final byte[] MAGIC =
"V88-WAL".getBytes(StandardCharsets.US_ASCII);
+ private static volatile WALFormatProvider delegate;
+
+ @Override
+ public byte[] getMagic() {
+ return MAGIC.clone();
+ }
+
+ @Override
+ public boolean isEnabled() {
+ return delegate != null && delegate.isEnabled();
+ }
+
+ private WALFormatProvider implementation() throws IOException {
+ if (delegate == null) {
+ throw WALFormatException.unsupported();
+ }
+ return delegate;
+ }
+
+ @Override
+ public WALFileContext create(File file, FileChannel channel) throws
IOException {
+ ByteBuffer magic = ByteBuffer.wrap(MAGIC);
+ while (magic.hasRemaining()) {
+ channel.write(magic);
+ }
+ return implementation().create(file, channel);
+ }
+
+ @Override
+ public WALFileContext open(File file, FileChannel channel) throws
IOException {
+ channel.position(MAGIC.length);
+ return implementation().open(file, channel);
+ }
+
+ @Override
+ public WALMetaData readMetadata(File file, FileChannel channel) throws
IOException {
+ return implementation().readMetadata(file, channel);
+ }
+
+ @Override
+ public void validateRead(File file, WALFileVersion version) throws
IOException {
+ if (delegate != null) {
+ delegate.validateRead(file, version);
+ }
+ }
+
+ @Override
+ public void validateForRecovery(File file) throws IOException {
+ implementation().validateForRecovery(file);
+ }
+
+ @Override
+ public void onRecoveryFailure(Throwable failure) {
+ delegate.onRecoveryFailure(failure);
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatsTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatsTest.java
new file mode 100644
index 00000000000..cd36779ae90
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatsTest.java
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.mockito.InOrder;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class WALFormatsTest {
+ @Rule public final TemporaryFolder folder = new TemporaryFolder();
+
+ @Test
+ public void testProviderIsOptionalAndUnambiguous() {
+ assertNull(WALFormats.selectProvider(Collections.emptyIterator()));
+ WALFormatProvider provider = mock(WALFormatProvider.class);
+
when(provider.getMagic()).thenReturn("V88-WAL".getBytes(StandardCharsets.US_ASCII));
+ assertSame(provider,
WALFormats.selectProvider(Collections.singleton(provider).iterator()));
+ assertThrows(
+ IllegalStateException.class,
+ () -> WALFormats.selectProvider(Arrays.asList(provider,
provider).iterator()));
+ }
+
+ @Test
+ public void testProviderMagicIsBounded() {
+ for (byte[] magic :
+ new byte[][] {
+ null, new byte[0], new byte[65], new byte[] {1, 2, 3},
WALFileVersion.V3.getVersionBytes()
+ }) {
+ WALFormatProvider provider = mock(WALFormatProvider.class);
+ when(provider.getMagic()).thenReturn(magic);
+ assertThrows(
+ IllegalStateException.class,
+ () ->
WALFormats.selectProvider(Collections.singleton(provider).iterator()));
+ }
+ }
+
+ @Test
+ public void testUnknownVersionIsNotGuessedAsLegacy() throws Exception {
+ File file = folder.newFile();
+ Files.write(file.toPath(),
"V77-WALpayload".getBytes(StandardCharsets.US_ASCII));
+ try (FileChannel channel = FileChannel.open(file.toPath())) {
+ channel.position(2);
+ assertThrows(WALFormatException.class, () ->
WALFileVersion.getVersion(channel));
+ assertEquals(2, channel.position());
+ assertThrows(WALFormatException.class, () ->
WALFormats.recognizes(channel, null));
+ assertEquals(2, channel.position());
+ }
+ assertThrows(WALFormatException.class, () -> new WALInputStream(file));
+ }
+
+ @Test
+ public void testRecognitionDoesNotDependOnWritePolicy() throws Exception {
+ WALFormatProvider provider = mock(WALFormatProvider.class);
+ byte[] magic = "V88-WAL".getBytes(StandardCharsets.US_ASCII);
+ when(provider.getMagic()).thenReturn(magic);
+ when(provider.isEnabled()).thenReturn(false);
+ File file = folder.newFile();
+ Files.write(file.toPath(), magic);
+ try (FileChannel channel = FileChannel.open(file.toPath())) {
+ channel.position(1);
+ assertTrue(WALFormats.recognizes(channel, provider));
+ assertEquals(1, channel.position());
+ }
+ }
+
+ @Test
+ public void testPlainAndEmptyFilesDoNotRequireProvider() throws Exception {
+ File file = folder.newFile();
+ try (FileChannel channel = FileChannel.open(file.toPath())) {
+ assertFalse(WALFormats.recognizes(channel, null));
+ assertEquals(WALFileVersion.V1, WALFileVersion.getVersion(channel));
+ }
+ Files.write(file.toPath(), new byte[] {0, 1, 2});
+ try (FileChannel channel = FileChannel.open(file.toPath())) {
+ assertFalse(WALFormats.recognizes(channel, null));
+ assertEquals(WALFileVersion.V1, WALFileVersion.getVersion(channel));
+ }
+ Files.write(file.toPath(), WALFileVersion.V3.getVersionBytes());
+ try (FileChannel channel = FileChannel.open(file.toPath())) {
+ assertEquals(WALFileVersion.V3, WALFileVersion.getVersion(channel));
+ assertTrue(WALFileVersion.isEmptyOrHeaderOnly(channel));
+ }
+ }
+
+ @Test
+ public void testReaderReleasesBufferBeforeClosingFormatContext() throws
Exception {
+ File file = folder.newFile();
+ WALInputStream reader = new WALInputStream(file);
+ WALFileContext context = mock(WALFileContext.class);
+ ByteBuffer data = ByteBuffer.wrap(new byte[] {1, 2, 3});
+ Field contextField = WALInputStream.class.getDeclaredField("fileContext");
+ contextField.setAccessible(true);
+ contextField.set(reader, context);
+ Field dataField = WALInputStream.class.getDeclaredField("dataBuffer");
+ dataField.setAccessible(true);
+ dataField.set(reader, data);
+ reader.close();
+ InOrder order = inOrder(context);
+ order.verify(context).release(data);
+ order.verify(context).close();
+ }
+
+ @Test
+ public void testUnrecoverableFailureSurvivesWrappingWithoutLoopingOnCauses()
{
+ WALFormatException fatal = WALFormatException.unsupported();
+ assertTrue(WALFormatException.isFatal(new IllegalStateException(new
IOException(fatal))));
+ IOException first = new IOException();
+ IOException second = new IOException(first);
+ first.initCause(second);
+ assertFalse(WALFormatException.isFatal(first));
+ assertFalse(WALFormatException.isFatal(new IOException()));
+ }
+}
