diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 1c0362b627b3a..5e92afb5f9cf0 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/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: {}. {}"; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 49096659ade1b..14bc9cd4b266b 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/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 = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java index b692ec94ed046..8d2f0af7cec88 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java +++ b/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, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java index ee61c105c1507..6572473cb58c2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java +++ b/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( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java index 45bff39d6ba78..c800a2327e900 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java +++ b/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");