[IOTDB-6262] Pipe: fix NPE while deserializing WAL (caused by non-atomic WAL rename operation during pipe read)

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/AbstractWALBuffer.java

@@ -31,6 +31,8 @@
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.StandardCopyOption;
 import java.util.Arrays;
 
 public abstract class AbstractWALBuffer implements IWALBuffer {
@@ -48,6 +50,7 @@ public abstract class AbstractWALBuffer implements IWALBuffer {
   protected volatile long currentWALFileVersion;
   // current search index
   protected volatile long currentSearchIndex;
+
   // current wal file log writer
   // it's safe to use volatile here to make this reference thread-safe.
   @SuppressWarnings("squid:S3077")
@@ -106,11 +109,12 @@ protected File rollLogWriter(long searchIndex, WALFileStatus fileStatus) throws
               WALFileUtils.parseStartSearchIndex(lastName),
               fileStatus);
       File targetFile = SystemFileFactory.INSTANCE.getFile(logDirectory, targetName);
-      if (lastFile.renameTo(targetFile)) {
-        lastFile = targetFile;
-      } else {
-        logger.error("Fail to rename file {} to {}", lastName, targetName);
-      }
+      Files.move(
+          lastFile.toPath(),
+          targetFile.toPath(),
+          StandardCopyOption.REPLACE_EXISTING,
+          StandardCopyOption.ATOMIC_MOVE);
+      lastFile = targetFile;
     }
     // roll file
     long nextFileVersion = currentWALFileVersion + 1;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java

@@ -930,7 +930,7 @@ public File getLogDirectory() {
   }
 
   /** Get the .wal file starts with the specified version id */
-  public File getWALFile(long versionId) {
+  public File getWALFile(long versionId) throws FileNotFoundException {
     return WALFileUtils.getWALFile(logDirectory, versionId);
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryHandler.java

@@ -41,13 +41,16 @@ public class WALEntryHandler {
   private static final Logger logger = LoggerFactory.getLogger(WALEntryHandler.class);
 
   private long memTableId = -1;
+
   // cached value, null after this value is flushed to wal successfully
   @SuppressWarnings("squid:S3077")
   private volatile WALEntryValue value;
+
   // wal entry's position in the wal, valid after the value is flushed to wal successfully
   // it's safe to use volatile here to make this reference thread-safe.
   @SuppressWarnings("squid:S3077")
   private final WALEntryPosition walEntryPosition = new WALEntryPosition();
+
   // wal node, null when wal is disabled
   private WALNode walNode = null;
 
@@ -84,9 +87,14 @@ public void unpinMemTable() throws MemTablePinException {
   }
 
   public InsertNode getInsertNodeViaCacheIfPossible() {
-    return value instanceof InsertNode
-        ? (InsertNode) value
-        : walEntryPosition.readByteBufferOrInsertNodeViaCacheDirectly().getRight();
+    try {
+      return value instanceof InsertNode
+          ? (InsertNode) value
+          : walEntryPosition.readByteBufferOrInsertNodeViaCacheDirectly().getRight();
+    } catch (Exception e) {
+      logger.warn("Fail to get insert node via cache. {}", this, e);
+      throw e;
+    }
   }
 
   /**

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALFileUtils.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.storageengine.dataregion.wal.utils;
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.regex.Matcher;
@@ -73,12 +74,15 @@ public static File[] listAllWALFiles(File dir) {
   }
 
   /** Get the .wal file starts with the specified version id in the directory. */
-  public static File getWALFile(File dir, long versionId) {
+  public static File getWALFile(File dir, long versionId) throws FileNotFoundException {
     String filePrefix = WAL_FILE_PREFIX + versionId + FILE_NAME_SEPARATOR;
     File[] files =
         dir.listFiles((d, name) -> walFilenameFilter(d, name) && name.startsWith(filePrefix));
     if (files == null || files.length != 1) {
-      return null;
+      throw new FileNotFoundException(
+          String.format(
+              "Fail to get wal file by versionId=%s and files=%s.",
+              versionId, Arrays.toString(files)));
     }
     return files[0];
   }