[Fix] Preserve absent local search index during subscription WAL replay

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -2216,10 +2216,9 @@ private DataNodePipeMessages() {}
       "ProgressWALIterator: skipped {} unreadable retained WAL files in directory {}, "
           + "firstFile={}, lastFile={}, firstError={}; historical subscription data in these "
           + "files cannot be replayed";
-  public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64 =
-      "ConsensusPrefetchingQueue {}: WAL replay skipped unavailable search indexes [{}, {}), "
-          + "skippedEntries={}, totalWalGapSkippedEntries={}; the missing WAL data may have been "
-          + "reclaimed before subscription consumption";
+  public static final String
+      LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A =
+          "{}: WAL gap [{}, {}), skipped {}, total skipped {}; cause undetermined";
   public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 =
       "Pipe {}@{}: terminate event committed for historical transfer. creationTime: {}, "
           + "shouldMark: {}. {}";

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -2057,9 +2057,9 @@ private DataNodePipeMessages() {}
   public static final String PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E =
       "ProgressWALIterator：跳过了 {} 个无法读取的保留 WAL 文件，directory={}，firstFile={}，"
           + "lastFile={}，firstError={}；这些文件中的历史订阅数据无法重放";
-  public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64 =
-      "ConsensusPrefetchingQueue {}：WAL 重放跳过了不可用的 searchIndex 区间 [{}, {})，"
-          + "skippedEntries={}，totalWalGapSkippedEntries={}；缺失的 WAL 数据可能已在订阅消费前被回收";
+  public static final String
+      LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A =
+          "{}：WAL 缺口 [{}, {})，跳过 {} 条，累计跳过 {} 条；原因尚未确定";
   public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 =
       "Pipe {}@{}：历史传输的终止事件已提交。creationTime：{}，shouldMark：{}。{}";
   public static final String PIPE_LOG_PIPE_HISTORICAL_SOURCE_HAS_SUPPLIED_ALL_EVENTS_EMITTING_8B58DE19 =

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java

@@ -2043,7 +2043,7 @@ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest reque
       final long totalSkippedEntries = walGapSkippedEntries.addAndGet(skippedEntries);
       LOGGER.warn(
           DataNodePipeMessages
-              .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64,
+              .LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A,
           this,
           expectedSearchIndex,
           actualSearchIndex,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java

@@ -734,7 +734,9 @@ private long getSearchIndexOrMetadataFallback(
     }
     buffer.position(SEARCH_INDEX_OFFSET);
     final long bodySearchIndex = SearchNode.extractSearchIndex(buffer.getLong());
-    return bodySearchIndex >= 0 ? bodySearchIndex : metadataSearchIndex;
+    // A replicated request has no local search index. The metadata fallback is based on the
+    // WAL entry offset, which also counts fragments and cannot assign a local index to it.
+    return bodySearchIndex;
   }
 
   private void startPending(

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java

@@ -22,6 +22,7 @@
 import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
 import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode;
 import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
 import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
 import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion;
@@ -87,7 +88,7 @@ public void testIteratorGroupsByLocalSeqAndCarriesWriterMetadata() throws Except
   }
 
   @Test
-  public void testIteratorUsesMetadataSearchIndexForStartFiltering() throws Exception {
+  public void testIteratorUsesBodySearchIndexForStartFiltering() throws Exception {
     final Path dir = Files.createTempDirectory("progress-wal-iterator-metadata-search-index");
     final File firstWal =
         dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX))
@@ -98,8 +99,8 @@ public void testIteratorUsesMetadataSearchIndexForStartFiltering() throws Except
 
     try {
       try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) {
-        writer.write(searchableEntry(-1L), singleEntryMeta(19, 5L, 1L, 1000L, 7, 105L));
-        writer.write(searchableEntry(-1L), singleEntryMeta(19, 6L, 1L, 2000L, 7, 106L));
+        writer.write(searchableEntry(5L), singleEntryMeta(19, 5L, 1L, 1000L, 7, 105L));
+        writer.write(searchableEntry(6L), singleEntryMeta(19, 6L, 1L, 2000L, 7, 106L));
       }
       try (WALWriter ignored = new WALWriter(lastWal, WALFileVersion.V3)) {
         // Create a sealed successor so the first WAL becomes historical and readable.
@@ -121,6 +122,75 @@ public void testIteratorUsesMetadataSearchIndexForStartFiltering() throws Except
     }
   }
 
+  @Test
+  public void testIteratorPreservesAbsentLocalIndexAfterManyFragments() throws Exception {
+    final Path dir = Files.createTempDirectory("progress-wal-iterator-mixed-fragments");
+    final File dataWal =
+        dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX))
+            .toFile();
+    final File successorWal =
+        dir.resolve(WALFileUtils.getLogFileName(1, 1878, WALFileStatus.CONTAINS_SEARCH_INDEX))
+            .toFile();
+
+    try {
+      try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) {
+        // Physical WAL entries count fragments, not local consensus requests. The old fallback
+        // assigns the replicated entry 1877 + 550 = 2427, creating a false [1878, 2427) gap.
+        for (int fragment = 0; fragment < 550; fragment++) {
+          writer.write(searchableEntry(1877L), singleEntryMeta(19, 1877L, 1L, 1000L, 7, 1877L));
+        }
+        writer.write(
+            searchableEntry(SearchNode.NO_CONSENSUS_INDEX),
+            singleEntryMeta(19, SearchNode.NO_CONSENSUS_INDEX, 1L, 2000L, 8, 10000L));
+        writer.write(searchableEntry(1878L), singleEntryMeta(19, 1878L, 1L, 3000L, 7, 1878L));
+      }
+      try (WALWriter ignored = new WALWriter(successorWal, WALFileVersion.V3)) {
+        // Keep both files retained and sealed; there is no deletion or concurrent write.
+      }
+
+      try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) {
+        assertTrue(iterator.hasNext());
+        final IndexedConsensusRequest fragmented = iterator.next();
+        assertEquals(1877L, fragmented.getSearchIndex());
+        assertEquals(1877L, fragmented.getProgressLocalSeq());
+        assertEquals(7, fragmented.getNodeId());
+        assertEquals(550, fragmented.getRequests().size());
+
+        assertTrue(iterator.hasNext());
+        final IndexedConsensusRequest replicated = iterator.next();
+        assertEquals(SearchNode.NO_CONSENSUS_INDEX, replicated.getSearchIndex());
+        assertEquals(10000L, replicated.getProgressLocalSeq());
+        assertEquals(2000L, replicated.getPhysicalTime());
+        assertEquals(8, replicated.getNodeId());
+        assertEquals(1, replicated.getRequests().size());
+
+        assertTrue(iterator.hasNext());
+        final IndexedConsensusRequest nextLocal = iterator.next();
+        assertEquals(1878L, nextLocal.getSearchIndex());
+        assertEquals(1878L, nextLocal.getProgressLocalSeq());
+        assertEquals(7, nextLocal.getNodeId());
+        assertEquals(1, nextLocal.getRequests().size());
+        assertFalse(iterator.hasNext());
+        assertFalse(iterator.hasIncompleteScan());
+        assertTrue(dataWal.isFile());
+      }
+
+      // A local-index seek must also retain replicated requests with their own writer progress.
+      try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1878L)) {
+        assertTrue(iterator.hasNext());
+        assertEquals(SearchNode.NO_CONSENSUS_INDEX, iterator.next().getSearchIndex());
+        assertTrue(iterator.hasNext());
+        assertEquals(1878L, iterator.next().getSearchIndex());
+        assertFalse(iterator.hasNext());
+        assertFalse(iterator.hasIncompleteScan());
+      }
+    } finally {
+      Files.deleteIfExists(dataWal.toPath());
+      Files.deleteIfExists(successorWal.toPath());
+      Files.deleteIfExists(dir);
+    }
+  }
+
   @Test
   public void testIteratorMergesFragmentsWithSameLocalSeq() throws Exception {
     final Path dir = Files.createTempDirectory("progress-wal-iterator-merge");