[Storage] Add optional WAL and checkpoint format extension interfaces

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={}.";

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={}。";

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.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 @@ private void asyncRecoverDataRegion() throws StartupException {
         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;
+        }
       }
     }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointReader.java

@@ -29,6 +29,7 @@
 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,13 +54,18 @@ private void init() {
         Checkpoint checkpoint = Checkpoint.deserialize(logStream);
         checkpoints.add(checkpoint);
       }
+    } catch (WALFormatException e) {
+      throw new UncheckedIOException(e);
     } catch (EOFException e) {
       logger.debug(
           StorageEngineMessages
               .STORAGE_LOG_MEET_ERROR_WHEN_READING_CHECKPOINT_FILE_SKIP_BROKEN_CHECKPOINTS_DADF9E9D,
           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,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointWriter.java

@@ -27,6 +27,6 @@
 /** 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));
   }
 }

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,21 +71,31 @@ protected LogWriter(File logFile, WALFileVersion version) throws IOException {
             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 {
         logChannel.close();
       } 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 double write(ByteBuffer buffer, boolean allowCompress) throws IOException
       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 void close() throws IOException {
           logChannel.force(true);
         }
       } finally {
-        logChannel.close();
+        try {
+          logChannel.close();
+        } finally {
+          if (fileContext != null) {
+            fileContext.close();
+          }
+        }
       }
     }
   }

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();
+}

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 @@
 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 static WALFileVersion getVersion(FileChannel channel) throws IOException
           return version;
         }
       }
+      if (WALFormats.recognizes(channel)) {
+        return EXTENDED;
+      }
       // v1 by default
       return V1;
     } finally {

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;
+  }
+}

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) {}
+}