(iotdb) branch master updated: Skip broken wal files when iterate the PlanNodeIterator (#11935)

[tanxinyu]

This is an automated email from the ASF dual-hosted git repository.

tanxinyu 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 db7d115204a Skip broken wal files when iterate the PlanNodeIterator 
(#11935)
db7d115204a is described below

commit db7d115204a8026ec1cde25ffbbaebb7542d4a63
Author: Alan Choo <43991780+heimi...@users.noreply.github.com>
AuthorDate: Sun Jan 21 16:23:12 2024 +0800

    Skip broken wal files when iterate the PlanNodeIterator (#11935)
---
 .../db/storageengine/dataregion/wal/node/WALNode.java     | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
index 9501834f475..9d117c6dccf 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java
@@ -630,6 +630,8 @@ public class WALNode implements IWALNode {
     private final LinkedList<IndexedConsensusRequest> insertNodes = new 
LinkedList<>();
     /** iterator of insertNodes */
     private ListIterator<IndexedConsensusRequest> itr = null;
+    /** last broken wal file's version id */
+    private long brokenFileId = -1;
 
     public PlanNodeIterator(long startIndex) {
       this.nextSearchIndex = startIndex;
@@ -644,9 +646,12 @@ public class WALNode implements IWALNode {
       // clear outdated iterator
       insertNodes.clear();
       itr = null;
+      if (filesToSearch == null || currentFileIndex >= filesToSearch.length - 
1) {
+        needUpdatingFilesToSearch = true;
+      }
 
       // update files to search
-      if (needUpdatingFilesToSearch || filesToSearch == null) {
+      if (needUpdatingFilesToSearch) {
         updateFilesToSearch();
         if (needUpdatingFilesToSearch) {
           logger.debug(
@@ -706,6 +711,7 @@ public class WALNode implements IWALNode {
         reset();
         return hasNext();
       } catch (Exception e) {
+        brokenFileId = 
WALFileUtils.parseVersionId(filesToSearch[currentFileIndex].getName());
         logger.error(
             "Fail to read wal from wal file {}, skip this file.",
             filesToSearch[currentFileIndex],
@@ -769,6 +775,7 @@ public class WALNode implements IWALNode {
             reset();
             return hasNext();
           } catch (Exception e) {
+            brokenFileId = 
WALFileUtils.parseVersionId(filesToSearch[fileIndex].getName());
             logger.error(
                 "Fail to read wal from wal file {}, skip this file.", 
filesToSearch[fileIndex], e);
           }
@@ -874,6 +881,7 @@ public class WALNode implements IWALNode {
       itr = null;
       filesToSearch = null;
       currentFileIndex = -1;
+      brokenFileId = -1;
       needUpdatingFilesToSearch = true;
     }
 
@@ -888,6 +896,11 @@ public class WALNode implements IWALNode {
       if (fileIndex == -1) {
         fileIndex = 0;
       }
+      // skip broken files
+      while (fileIndex < filesToSearch.length - 1
+          && WALFileUtils.parseVersionId(filesToSearch[fileIndex].getName()) 
<= brokenFileId) {
+        fileIndex++;
+      }
       if (filesToSearch != null
           && (fileIndex >= 0 && fileIndex < filesToSearch.length - 1)) { // 
possible to find next
         this.filesToSearch = filesToSearch;